Declaración de una corriente de RabbitMQ

Puede crear una corriente utilizando la interfaz de gestión de RabbitMQ.

• En primer lugar, seleccione Añadir una cola nueva.
• A continuación, en el desplegable Tipo, seleccione Corriente.

De forma alternativa, cree una corriente utilizando la interfaz de gestión de RabbitMQ creando una cola con el tipo “stream”. Si una cola ya está presente, no se creará. Declare una corriente con un mandato como:

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"}
)

Publicación de una corriente de RabbitMQ

El script siguiente declara una ruta de RabbitMQ (test_stream) y, a continuación, publica un mensaje en ella a través de la función 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'
)

Consumo de una corriente de RabbitMQ

Los mensajes se pueden consumir desde una secuencia de la misma forma que las colas realizan esta tarea, con dos diferencias principales:

1. El consumo de mensajes en RabbitMQ Streams requiere establecer la captación previa deQoS.
2. Especifique un desplazamiento para empezar a leer/consumir desde cualquier punto de la secuencia de registro. Si no se especifica, el consumidor empieza a leer el desplazamiento más reciente grabado en la secuencia de registro después de que se inicie.

El script siguiente declara test_stream de nuevo, estableciendo la captación previa QoS en 100 utilizando la función basic_qos. El consumidor desencadena la función de devolución de llamada cuando procesa un mensaje nuevo. A continuación, la función de devolución de llamada invoca la función send_welcome_email que simula el envío de un correo electrónico a un usuario.

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()

Tenga en cuenta cómo no se especifica un desplazamiento en nuestro basic_consume: # Consume messages published to the stream channel.basic_consume( 'test_stream', callback). Como resultado, el consumidor empieza a leer desde el desplazamiento más reciente escrito en test_stream después de que se inicie el consumidor. Después de se ha enfatizado deliberadamente aquí para permitir el examen cruzado de un comportamiento interesante de las rutas.