声明 RabbitMQ 流

您可以使用 RabbitMQ 管理界面来创建流。

• 首先，选择 添加新队列。
• 接下来，在 类型 下拉列表中，选择 流。

或者，通过使用 RabbitMQ 管理界面创建类型为 “stream” 的队列来创建流。 如果队列已存在，那么将不会创建该队列。 使用类似如下的命令声明流:

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 流

可以通过队列完成此任务的相同方式从流中使用消息，这有两个主要差异:

1. 在 RabbitMQ Streams 中使用消息需要设置 QoS 预取。
2. 指定从日志流中的任何点开始读取/使用的偏移量。 如果未指定，那么使用者将在日志流启动后开始从写入日志流的最新偏移量中进行读取。

以下脚本再次声明 test_stream，并使用 basic_qos 函数将 QoS 预取设置为 100。 使用者在处理新消息时触发回调函数。 然后，回调函数将调用模拟向用户发送电子邮件的 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 的最新偏移量开始读取。 此处特意强调了 之后，以允许对流的有趣行为进行交叉检查。