Déclaration d'un flux RabbitMQ

Vous pouvez créer un flux à l'aide de l'interface de gestion RabbitMQ.

• Sélectionnez d'abord Ajouter une nouvelle file d'attente.
• Ensuite, dans la liste déroulante Type, sélectionnez Flux.

Vous pouvez également créer un flux à l'aide de l'interface de gestion RabbitMQ en créant une file d'attente de type “stream”. Si une file d'attente est déjà présente, elle ne sera pas créée. Déclarez un flux avec une commande telle que:

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

Publication d'un flux RabbitMQ

Le script suivant déclare un flux RabbitMQ (test_stream), puis y publie un message via la fonction 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'
)

Consommation d'un flux RabbitMQ

Les messages peuvent être consommés à partir d'un flux de la même manière que les files d'attente effectuent cette tâche, avec deux différences majeures:

1. La consommation de messages dans RabbitMQ Streams nécessite la définition de la préextractionQoS.
2. Indiquez un décalage pour commencer à lire / consommer à partir de n'importe quel point du flux de journalisation. S'il n'est pas spécifié, le consommateur commence la lecture à partir du décalage le plus récent écrit dans le flux de journalisation après son démarrage.

Le script suivant déclare à nouveau test_stream, en définissant la préextraction QoS sur 100 à l'aide de la fonction basic_qos. Le consommateur déclenche la fonction de rappel lorsqu'il traite un nouveau message. La fonction de rappel appelle ensuite la fonction send_welcome_email qui simule l'envoi d'un courrier électronique à un utilisateur.

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

Notez qu'un décalage n'est pas spécifié dans notre basic_consume: # Consume messages published to the stream channel.basic_consume( 'test_stream', callback). Par conséquent, le consommateur commence à lire à partir du décalage le plus récent écrit dans test_stream après le démarrage du consommateur. Après a été délibérément souligné ici pour permettre le contre-interrogatoire d'un comportement intéressant des flux.