RabbitMQ ストリームの宣言

RabbitMQ 管理インターフェースを使用してストリームを作成できます。

• 最初に、 「新規キューの追加」 を選択します。
• 次に、 「タイプ」 ドロップダウンで *「ストリーム」*を選択します。

あるいは、タイプ “stream” のキューを作成して、 RabbitMQ 管理インターフェースを使用してストリームを作成します。 キューが既に存在する場合、そのキューは作成されません。 以下のようなコマンドを使用してストリームを宣言します。

import pika, os

credentials = pika.PlainCredentials('username', 'password')
context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
context.verify_mode = ssl.CERT_NONE
ssl_options = pika.SSLOptions(context)

# Connection
connection = pika.BlockingConnection(
  pika.ConnectionParameters(
    host='<hostname>',
    port='<port>',
    credentials=credentials,
    virtual_host="/",
    ssl_options=ssl_options))
channel = connection.channel() # start a channel

# Declare a Stream, named test_stream
channel.queue_declare(
  queue='test_stream',
      durable=True,
  arguments={"x-queue-type": "stream"}
)

RabbitMQ ストリームのパブリッシュ

以下のスクリプトは、 RabbitMQ ストリーム (test_stream) を宣言し、 basic_publish 関数を介してそのストリームにメッセージをパブリッシュします。

import pika, os

credentials = pika.PlainCredentials('username', 'password')
context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
context.verify_mode = ssl.CERT_NONE
ssl_options = pika.SSLOptions(context)

# Connection
connection = pika.BlockingConnection(
  pika.ConnectionParameters(
    host='<hostname>',
    port='<port>',
    credentials=credentials,
    virtual_host="/",
    ssl_options=ssl_options))
channel = connection.channel() # start a channel

# Declare a Stream, named test_stream
channel.queue_declare(
  queue='test_stream',
      durable=True,
  arguments={"x-queue-type": "stream"}
)

# Publish a message to the test_stream
channel.basic_publish(
  exchange='',
  routing_key='test_stream',
  body='Welcome email message'
)

RabbitMQ ストリームのコンシューム

メッセージは、キューがこのタスクを実行するのと同じ方法でストリームから取り込むことができますが、以下の 2 つの大きな違いがあります。

1. RabbitMQ Streams でメッセージをコンシュームするには、 QoS プリフェッチを設定する必要があります。
2. ログ・ストリーム内の任意のポイントから読み取り/消費を開始するオフセットを指定します。 指定されていない場合、コンシューマーは、開始後にログ・ストリームに書き込まれた最新のオフセットから読み取りを開始します。

以下のスクリプトは、 test_stream を再度宣言し、 basic_qos 関数を使用して QoS プリフェッチを 100 に設定します。 コンシューマーは、新規メッセージの処理時にコールバック関数をトリガーします。 次に、コールバック関数は、ユーザーへの E メールの送信をシミュレートする send_welcome_email 関数を呼び出します。

import pika, os, time

def send_welcome_email(msg):
  print("Welcome Email task processing")
  print(" [x] Received " + str(msg))
  time.sleep(5) # simulate sending email to a user --delays for 5 seconds
  print("Email successfully sent!")
  return

credentials = pika.PlainCredentials('username', 'password')
context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
context.verify_mode = ssl.CERT_NONE
ssl_options = pika.SSLOptions(context)

# Connection
connection = pika.BlockingConnection(
  pika.ConnectionParameters(
    host='<hostname>',
    port='<password>',
    credentials=credentials,
    virtual_host="/",
    ssl_options=ssl_options))
channel = connection.channel() # start a channel

# Declare our stream
channel.queue_declare(
  queue='test_stream',
  durable=True,
  arguments={"x-queue-type": "stream"}
)

# create a function which is called on incoming messages
def callback(ch, method, properties, body):
  send_welcome_email(body)

# Set the consumer QoS prefetch
channel.basic_qos(
  prefetch_count=100
)

# Consume messages published to the stream
channel.basic_consume(
  'test_stream',
  callback,
)

# start consuming (blocks)
channel.start_consuming()
connection.close()

basic_consume でオフセットがどのように指定されていないかに注意してください: # Consume messages published to the stream channel.basic_consume( 'test_stream', callback)。 その結果、コンシューマーは、開始後に test_stream に書き込まれた最新のオフセットから読み取りを開始します。 ここでは、 後 を意図的に強調して、ストリームの関心のある振る舞いを相互に調査できるようにしています。