RabbitMQ-Datenstrom deklarieren

Sie können einen Datenstrom mithilfe der RabbitMQ-Managementschnittstelle erstellen.

• Wählen Sie zunächst Neue Warteschlange hinzufügen aus.
• Wählen Sie als Nächstes in der Dropdown-Liste Typ die Option Datenstromaus.

Alternativ können Sie einen Stream erstellen, indem Sie die RabbitMQ-Managementschnittstelle verwenden, indem Sie eine Warteschlange mit dem Typ “stream” erstellen. Wenn bereits eine Warteschlange vorhanden ist, wird sie nicht erstellt. Deklarieren Sie einen Datenstrom mit einem Befehl wie dem folgenden:

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-Datenstrom veröffentlichen

Das folgende Script deklariert einen RabbitMQ-Stream (test_stream) und veröffentlicht anschließend über die Funktion basic_publish eine Nachricht:

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-Datenstrom verarbeiten

Nachrichten können aus einem Stream auf dieselbe Weise konsumiert werden, wie Warteschlangen diese Task ausführen, mit zwei Hauptunterschieden:

1. Das Verarbeiten von Nachrichten in RabbitMQ Streams erfordert die Einstellung des QoS Vorablesezugriffs.
2. Geben Sie einen Offset an, um das Lesen/Konsumieren von einem beliebigen Punkt im Protokolldatenstrom zu starten. Wenn nicht angegeben, beginnt der Konsument mit dem Lesen des zuletzt in den Protokolldatenstrom geschriebenen Offsets, nachdem er gestartet wurde.

Das folgende Script deklariert test_stream erneut und setzt den QoS-Vorablesezugriff mit der Funktion basic_qos auf 100. Der Konsument löst die Callback-Funktion aus, wenn er eine neue Nachricht verarbeitet. Die Callback-Funktion ruft dann die Funktion send_welcome_email auf, die das Senden einer E-Mail an einen Benutzer simuliert.

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

Beachten Sie, dass in basic_consume kein Offset angegeben ist: # Consume messages published to the stream channel.basic_consume( 'test_stream', callback). Daher beginnt der Konsument mit dem Lesen des zuletzt in test_stream geschriebenen Offsets, nachdem der Konsument gestartet wurde. Nach wurde hier bewusst betont, um die Queruntersuchung eines interessanten Verhaltens von Streams zu ermöglichen.