RabbitMQ 6中工作模式（Python 版）

Hello World

20210815210502109.png

send.py

#!/usr/bin/env python
import pika
 
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
 
channel.queue_declare(queue='hello')
 
channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
print(" [x] Sent 'Hello World!'")
connection.close()

receive.py

#!/usr/bin/env python
import pika, sys, os
 
def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()
 
    channel.queue_declare(queue='hello')
 
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body.decode())
 
    channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)
 
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()
 
if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

python send.py
python receive.py

工作队列模式

20210815210534456.png

new_task.py

#!/usr/bin/env python
import pika
import sys
 
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
 
channel.queue_declare(queue='task_queue', durable=True)
 
message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(
    exchange='',
    routing_key='task_queue',
    body=message,
    properties=pika.BasicProperties(
        delivery_mode=2,  # make message persistent
    ))
print(" [x] Sent %r" % message)
connection.close()

worker.py

#!/usr/bin/env python
import pika
import time
 
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
 
channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
 
 
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body.decode())
    time.sleep(body.count(b'.'))
    print(" [x] Done")
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
 
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback)

python new_task.py "A very hard task which takes two seconds..."

python worker.py

发布 / 订阅模式

20210815210637479.png

receive_logs.py

#!/usr/bin/env python
import pika, sys, os
 
def main():
    connection = pika.BlockingConnection(
        pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()
 
    channel.exchange_declare(exchange='logs', exchange_type='fanout')
 
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
 
    channel.queue_bind(exchange='logs', queue=queue_name)
 
    def callback(ch, method, properties, body):
        print(" [x] %r" % body.decode())
 
    print(' [*] Waiting for logs. To exit press CTRL+C')
    channel.basic_consume(
        queue=queue_name, on_message_callback=callback, auto_ack=True)
 
    channel.start_consuming()
 
 
if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

emit_log.py

#!/usr/bin/env python
import pika
import sys
 
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='logs', exchange_type='fanout')
 
message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs', routing_key='', body=message)
print(" [x] Sent %r" % message)
connection.close()

python receive_logs.py

python emit_log.py "info: This is the log message"

路由模式

20210815210709554.png

recevie_logs_direct.py

#!/usr/bin/env python
import pika, sys, os
 
def main():
    connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()
 
    channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
 
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
 
    severities = sys.argv[1:]
    if not severities:
        sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
        sys.exit(1)
 
    for severity in severities:
        channel.queue_bind(
            exchange='direct_logs', queue=queue_name, routing_key=severity)
 
    print(' [*] Waiting for logs. To exit press CTRL+C')
 
 
    def callback(ch, method, properties, body):
        print(" [x] %r:%r" % (method.routing_key, body.decode()))
 
 
    channel.basic_consume(
        queue=queue_name, on_message_callback=callback, auto_ack=True)
 
    channel.start_consuming()
 
 
if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

emit_log_direct.py

#!/usr/bin/env python
import pika
import sys
 
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
 
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
 
severity = sys.argv[1] if len(sys.argv) > 2 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(
    exchange='direct_logs', routing_key=severity, body=message)
print(" [x] Sent %r:%r" % (severity, message))
connection.close()

python receive_logs_direct.py

python emit_log_direct.py info "The message"

主题模式

20210815210743108.png

receive_logs_topic.py

#!/usr/bin/env python
import pika, sys, os
 
def main():
    connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()
 
    channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
 
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
 
    binding_keys = sys.argv[1:]
    if not binding_keys:
        sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
        sys.exit(1)
 
    for binding_key in binding_keys:
        channel.queue_bind(
            exchange='topic_logs', queue=queue_name, routing_key=binding_key)
 
    print(' [*] Waiting for logs. To exit press CTRL+C')
 
 
    def callback(ch, method, properties, body):
        print(" [x] %r:%r" % (method.routing_key, body.decode()))
 
 
    channel.basic_consume(
        queue=queue_name, on_message_callback=callback, auto_ack=True)
 
    channel.start_consuming()
 
 
if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

python receive_logs_topic.py "*.rabbit"

python emit_log_topic.py red.rabbit Hello

RPC 模式

20210815210818177.png

rpc_server.py

#!/usr/bin/env python
import pika
 
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
 
channel = connection.channel()
 
channel.queue_declare(queue='rpc_queue')
 
 
def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)
 
 
def on_request(ch, method, props, body):
    n = int(body)
 
    print(" [.] fib(%s)" % n)
    response = fib(n)
 
    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id = \
                                                         props.correlation_id),
                     body=str(response))
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
 
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)
 
print(" [x] Awaiting RPC requests")
channel.start_consuming()

rpc_client.py

#!/usr/bin/env python
import pika
import uuid
 
 
class FibonacciRpcClient(object):
 
    def __init__(self):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host='localhost'))
 
        self.channel = self.connection.channel()
 
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue
 
        self.channel.basic_consume(
            queue=self.callback_queue,
            on_message_callback=self.on_response,
            auto_ack=True)
 
    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body
 
    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(
            exchange='',
            routing_key='rpc_queue',
            properties=pika.BasicProperties(
                reply_to=self.callback_queue,
                correlation_id=self.corr_id,
            ),
            body=str(n))
        while self.response is None:
            self.connection.process_data_events()
        return int(self.response)
 
 
fibonacci_rpc = FibonacciRpcClient()
 
print(" [x] Requesting fib(30)")
response = fibonacci_rpc.call(30)
print(" [.] Got %r" % response)

python rpc_server.py

python rpc_client.py