Dichiarazione di un flusso RabbitMQ

È possibile creare un flusso utilizzando l'interfaccia di gestione RabbitMQ.

• Innanzitutto, selezionare Aggiungi una nuova coda.
• Successivamente, nell'elenco a discesa Tipo, selezionare Flusso.

In alternativa, creare un flusso utilizzando l'interfaccia di gestione di RabbitMQ creando una coda con tipo “stream”. Se una coda è già presente, non verrà creata. Dichiarare un flusso con un comando come:

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

Pubblicazione di un flusso RabbitMQ

Il seguente script dichiara un flusso RabbitMQ (test_stream), quindi pubblica un messaggio attraverso la funzione 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'
)

Utilizzo di uno stream RabbitMQ

I messaggi possono essere utilizzati da un flusso nello stesso modo in cui le code eseguono questa attività, con due differenze principali:

1. L'utilizzo dei messaggi in RabbitMQ Streams richiede l'impostazione della ricercaQoS.
2. Specificare un offset per iniziare la lettura / utilizzo da qualsiasi punto del flusso di log. Se non specificato, il consumer inizia la lettura dall'offset più recente scritto nel flusso di log dopo l'avvio.

Il seguente script dichiara nuovamente test_stream, impostando la ricerca QoS su 100 utilizzando la funzione basic_qos. Il consumer attiva la funzione di callback quando elabora un nuovo messaggio. La funzione di callback richiama quindi la funzione send_welcome_email che simula l'invio di un'email a un utente.

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

Notare come un offset non è specificato in basic_consume: # Consume messages published to the stream channel.basic_consume( 'test_stream', callback). Di conseguenza, il consumer inizia a leggere dall'offset più recente scritto in test_stream dopo l'avvio del consumer. Dopo è stato deliberatamente enfatizzato qui per consentire l'esame incrociato di un comportamento interessante dei flussi.