Verwenden von Python Pika mit Amazon MQ for RabbitMQ

Das folgende Tutorial zeigt, wie Sie einen Python-Pika-Client mit TLS einrichten können, der für die Verbindung zu einem Amazon-MQ-for-RabbitMQ-Broker konfiguriert ist. Pika ist eine Python-Implementierung des AMQP-0-9-1-Protokolls für RabbitMQ. Dieses Tutorial führt Sie durch die Installation von Pika, das Deklarieren einer Warteschlange, das Einrichten eines Herausgebers für das Senden von Nachrichten an den Standardaustausch des Brokers und das Einrichten eines Verbrauchers, der Nachrichten aus der Warteschlange erhält.

Voraussetzungen

Um die Schritte dieses Tutorials auszuführen, benötigen Sie Folgendes:

• Einen Amazon-MQ-for-RabbitMQ-Broker. Weitere Informationen finden Sie unter Erstellen eines Amazon-MQ-for-RabbitMQ-Brokers.

• Python 3 für Ihr Betriebssystem installieren.

• Pika mithilfe von Python pip installiert. Öffnen Sie zum Installieren von Pika ein neues Terminalfenster und führen Sie Folgendes aus.

  $ python3 -m pip install pika

Berechtigungen

Für dieses Tutorial benötigen Sie mindestens einen Amazon-MQ-for-RabbitMQ-Brokerbenutzer mit der Berechtigung, an einen Vhost zu schreiben und von ihm zu lesen. Die folgende Tabelle enthält die erforderlichen Mindestberechtigungen als reguläre Ausdrücksmuster (regexp).

Tags (Markierungen)	Konfigurieren von regexp	REGEXP-Schreiben	Lesen Sie regexp
none		.*	.*

Die aufgelisteten Benutzerberechtigungen bieten dem Benutzer nur Lese- und Schreibberechtigungen, ohne Zugriff auf das Management-Plug-In zu gewähren, um Verwaltungsvorgänge für den Broker auszuführen. Sie können Berechtigungen weiter einschränken, indem Sie regexp-Muster bereitstellen, die den Zugriff des Benutzers auf bestimmte Warteschlangen einschränken. Zum Beispiel, wenn Sie das Lese-regexp-Muster auf ^[hello world].* ändern, hat der Benutzer nur die Berechtigung, aus Warteschlangen zu lesen, die mit hello world starten.

Weitere Informationen zum Erstellen von RabbitMQ-Benutzern und zum Verwalten von Benutzer-Tags und -Berechtigungen finden Sie unterBenutzer.

Schritt eins: Erstellen Sie einen einfachen Python-Pika-Client

Um eine Python-Pika-Client-Basisklasse zu erstellen, die einen Konstruktor definiert und den SSL-Kontext bereitstellt, der für die TLS-Konfiguration erforderlich ist, wenn Sie mit einem Amazon-MQ-for-RabbitMQ-Broker interagieren, machen Sie folgendes.

1. Öffnen Sie ein neues Terminalfenster, erstellen Sie ein neues Verzeichnis für Ihr Projekt und navigieren Sie zum Verzeichnis.

   $ mkdir pika-tutorial
   $ cd pika-tutorial

2. Erstellen Sie eine neue Datei, basicClient.py, die folgenden Python-Code enthält.

   import ssl
   import pika
   
   class BasicPikaClient:
   
       def __init__(self, rabbitmq_broker_id, rabbitmq_user, rabbitmq_password, region):
   
           # SSL Context for TLS configuration of Amazon MQ for RabbitMQ
           ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
           ssl_context.set_ciphers('ECDHE+AESGCM:!ECDSA')
   
           url = f"amqps://{rabbitmq_user}:{rabbitmq_password}@{rabbitmq_broker_id}.mq.{region}.amazonaws.com:5671"
           parameters = pika.URLParameters(url)
           parameters.ssl_options = pika.SSLOptions(context=ssl_context)
   
           self.connection = pika.BlockingConnection(parameters)
           self.channel = self.connection.channel()

Sie können jetzt zusätzliche Klassen für Ihren Herausgeber und Verbraucher definieren, die von BasicPikaClient erben.

Schritt zwei: Erstellen Sie einen Herausgeber und senden Sie eine Nachricht

Gehen Sie wie folgt vor, um einen Herausgeber zu erstellen, der eine Warteschlange deklariert und eine einzelne Nachricht sendet.

1. Kopieren Sie den Inhalt des folgenden Codebeispiels und speichern Sie es lokal als publisher.py im selben Verzeichnis, das Sie im vorherigen Schritt erstellt haben.

   from basicClient import BasicPikaClient
   
   class BasicMessageSender(BasicPikaClient):
   
       def declare_queue(self, queue_name):
           print(f"Trying to declare queue({queue_name})...")
           self.channel.queue_declare(queue=queue_name)
   
       def send_message(self, exchange, routing_key, body):
           channel = self.connection.channel()
           channel.basic_publish(exchange=exchange,
                                 routing_key=routing_key,
                                 body=body)
           print(f"Sent message. Exchange: {exchange}, Routing Key: {routing_key}, Body: {body}")
   
       def close(self):
           self.channel.close()
           self.connection.close()
   
   if __name__ == "__main__":
   
       # Initialize Basic Message Sender which creates a connection
       # and channel for sending messages.
       basic_message_sender = BasicMessageSender(
           "<broker-id>",
           "<username>",
           "<password>",
           "<region>"
       )
   
       # Declare a queue
       basic_message_sender.declare_queue("hello world queue")
   
       # Send a message to the queue.
       basic_message_sender.send_message(exchange="", routing_key="hello world queue", body=b'Hello World!')
   
       # Close connections.
       basic_message_sender.close()

   Die BasicMessageSender-Klasse erbt von BasicPikaClient und implementiert zusätzliche Methoden zum Deklarieren einer Warteschlange, zum Senden einer Nachricht an die Warteschlange und zum Schließen von Verbindungen. Das Codebeispiel leitet eine Nachricht an den Standardaustausch weiter, wobei ein Routing-Schlüssel dem Namen der Warteschlange entspricht.

2. Unter if __name__ == "__main__":, ersetzen Sie die Parameter, die an die BasicMessageSender-constructor-Anweisung weitergegeben werden mit den folgenden Informationen.

   • <broker-id> - Die eindeutige ID, die Amazon MQ für die Broker-Instance generiert. Sie können die ID von Ihrem Broker ARN analysieren. Beispielsweise angesichts der folgenden ARN, arn:aws:mq:us-east-2:123456789012:broker:MyBroker:b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9, wäre die Broker-ID b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9.

   • <username> – Der Benutzername für einen Broker-Benutzer mit ausreichenden Berechtigungen zum Schreiben von Nachrichten an den Broker.

   • <password> – Das Passwort für einen Broker-Benutzer mit ausreichenden Berechtigungen zum Schreiben von Nachrichten an den Broker.

   • <region> – DieAWS-Region, in der Sie Ihren Amazon-MQ-for-RabbitMQ-Broker erstellt haben. Zum Beispiel us-west-2.

3. Führen Sie den folgenden Befehl im selben Verzeichnis aus, in dem Sie publisher.py erstellt haben.

   $ python3 publisher.py

   Wenn der Code erfolgreich ausgeführt wird, wird die folgende Meldung in Ihrem Terminalfenster angezeigt.

   Trying to declare queue(hello world queue)...
   Sent message. Exchange: , Routing Key: hello world queue, Body: b'Hello World!'

Schritt drei: Erstellen Sie einen Verbraucher und erhalten Sie eine Nachricht

Gehen Sie wie folgt vor, um einen Verbraucher zu erstellen, der eine einzelne Nachricht aus der Warteschlange erhält.

1. Kopieren Sie den Inhalt des folgenden Codebeispiels und speichern Sie es lokal als consumer.py im selben Verzeichnis.

   from basicClient import BasicPikaClient
   
   class BasicMessageReceiver(BasicPikaClient):
   
       def get_message(self, queue):
           method_frame, header_frame, body = self.channel.basic_get(queue)
           if method_frame:
               print(method_frame, header_frame, body)
               self.channel.basic_ack(method_frame.delivery_tag)
               return method_frame, header_frame, body
           else:
               print('No message returned')
   
       def close(self):
           self.channel.close()
           self.connection.close()
   
   
   if __name__ == "__main__":
   
       # Create Basic Message Receiver which creates a connection
       # and channel for consuming messages.
       basic_message_receiver = BasicMessageReceiver(
           "<broker-id>",
           "<username>",
           "<password>",
           "<region>"
       )
   
       # Consume the message that was sent.
       basic_message_receiver.get_message("hello world queue")
   
       # Close connections.
       basic_message_receiver.close()

   Ähnlich wie bei dem Herausgeber, den Sie im vorherigen Schritt erstellt haben, erbt BasicMessageReciever von BasicPikaClient und implementiert zusätzliche Methoden zum Empfangen einer einzelnen Nachricht und zum Schließen von Verbindungen.

2. In der if __name__ == "__main__":-Anweisung, ersetzen Sie die Parameter, die an den BasicMessageReciever-Constructor weitergegeben werden mit Ihren Informationen.

3. Führen Sie den folgenden Befehl in Ihrem Projektverzeichnis aus.

   $ python3 consumer.py

   Wenn der Code erfolgreich ausgeführt wird, werden der Nachrichtentext und die Header einschließlich des Routing-Schlüssels in Ihrem Terminalfenster angezeigt.

   <Basic.GetOk(['delivery_tag=1', 'exchange=', 'message_count=0', 'redelivered=False', 'routing_key=hello world queue'])> <BasicProperties> b'Hello World!'

Schritt vier: (Optional) Richten Sie eine Ereignisschleife ein und konsumieren Sie Nachrichten

Um mehrere Nachrichten aus einer Warteschlange zu konsumieren, verwenden Sie Pikas basic_consume-Methode und eine Callback-Funktion wie nachfolgend dargestellt

1. In consumer.py, fügen Sie die folgende Methodendefinition zur BasicMessageReceiver-Klasse hinzu.

   def consume_messages(self, queue):
       def callback(ch, method, properties, body):
           print(" [x] Received %r" % body)
   
       self.channel.basic_consume(queue=queue, on_message_callback=callback, auto_ack=True)
   
       print(' [*] Waiting for messages. To exit press CTRL+C')
       self.channel.start_consuming()
                   

2. In consumer.py, unter if __name__ == "__main__":, rufen Sie die consume_messages-Methode auf, die Sie im vorherigen Schritt definiert haben.

   if __name__ == "__main__":
   
       # Create Basic Message Receiver which creates a connection and channel for consuming messages.
       basic_message_receiver = BasicMessageReceiver(
           "<broker-id>",
           "<username>",
           "<password>",
           "<region>"
       )
   
       # Consume the message that was sent.
       # basic_message_receiver.get_message("hello world queue")
   
       # Consume multiple messages in an event loop.
       basic_message_receiver.consume_messages("hello world queue")
   
       # Close connections.
       basic_message_receiver.close()

3. Führen Sie consumer.py erneut aus, und falls dies erfolgreich ist, werden die Nachrichten in der Warteschlange in Ihrem Terminalfenster angezeigt.

   [*] Waiting for messages. To exit press CTRL+C
   [x] Received b'Hello World!'
   [x] Received b'Hello World!'
   ...

Als nächstes

• Weitere Informationen zu anderen unterstützten RabbitMQ-Clientbibliotheken finden Sie in der RabbitMQ-Client-Dokumentation auf der RabbitMQ-Website.