RabbitMQ용 Amazon MQ와 함께 Python Pika 사용

새 터미널 창을 열고 프로젝트에 대한 새 디렉터리를 생성한 다음, 해당 디렉터리로 이동합니다.

$ mkdir pika-tutorial
$ cd pika-tutorial

다음의 Python 코드가 포함된 basicClient.py라는 이름의 파일을 생성합니다.

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

다음 코드 샘플의 내용을 복사하고 로컬의 이전 단계에서 생성한 것과 동일한 디렉터리에 publisher.py로 저장합니다.

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

BasicMessageSender 클래스는 BasicPikaClient에서 상속하여 대기열을 선언하고, 대기열에 메시지를 전송하고, 연결을 닫는 추가 메서드를 구현합니다. 코드 샘플은 대기열 이름과 동일한 라우팅 키를 사용하여 메시지를 기본 교환으로 라우팅합니다.

if __name__ == "__main__": 아래에서 다음 정보가 포함된 BasicMessageSender 생성자 문으로 전달되는 파라미터를 교체합니다.

publisher.py를 생성한 것과 동일한 디렉터리에서 다음 명령을 실행합니다.

$ python3 publisher.py

코드가 성공적으로 실행되면 터미널 창에 다음과 같이 출력됩니다.

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

다음 코드 샘플의 내용을 복사하고 로컬의 동일한 디렉터리에 consumer.py로 저장합니다.

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

이전 단계에서 생성한 게시자와 유사하게, BasicMessageReciever는 BasicPikaClient에서 상속하여 단일 메시지를 수신하고 연결을 닫는 추가 메서드를 구현합니다.

if __name__ == "__main__": 문 아래에서 BasicMessageReciever 생성자로 전달되는 파라미터를 사용자의 정보로 교체합니다.

프로젝트 디렉터리에서 다음 명령을 실행합니다.

$ python3 consumer.py

코드가 성공적으로 실행되면 메시지 본문과 라우팅 키를 포함한 헤더가 터미널 창에 표시됩니다.

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

consumer.py에서 BasicMessageReceiver 클래스에 다음 메서드 정의를 추가합니다.

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

consumer.py의 if __name__ == "__main__": 아래에서 이전 단계에서 정의한 consume_messages 메서드를 호출합니다.

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

consumer.py를 다시 실행하고, 성공하면 대기 중인 메시지가 터미널 창에 표시됩니다.

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