发布-订阅

参考 https://www.cnblogs.com/shenh/p/10497244.html

模式一：fanout

1. 不用指定routing_key
2. 消息会发给所有的订阅者
3. 需要先启动订阅者

# 发布者
import json
import pika

credentials = pika.PlainCredentials('root', 'root')  # mq的用户名、密码
connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='10.131.70.49', port=5672,  # mq的IP、端口
    virtual_host='/',
    credentials=credentials
))
channel = connection.channel()
channel.exchange_declare(
    exchange='python-text',  # 指定队列
    durable=True,  # 消息持久化
    exchange_type='fanout' # 发布的模式
)
for i in range(10):
    message = json.dumps({"Order":"1000%s"%i})
    channel.basic_publish(
        exchange='python-text',
        routing_key='',
        body=message,
    )
connection.close()

# 订阅者
import pika

credentials = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='10.131.70.49', port=5672, virtual_host='/', credentials=credentials))
channel = connection.channel()
# 创建临时队列，consumer关闭后，队列自动删除
result = channel.queue_declare('', exclusive=True)
# 声明exchange，由exchange指定消息在哪个队列传递，如不存在，则创建。durable = True 代表exchange持久化存储，False 非持久化存储
channel.exchange_declare(exchange='python-text', durable=True, exchange_type='fanout')
# 绑定exchange和队列  exchange 使我们能够确切地指定消息应该到哪个队列去
channel.queue_bind(exchange='python-text', queue=result.method.queue)


# 定义一个回调函数来处理消息队列中的消息，这里是打印出来
def callback(ch, method, properties, body):
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(body.decode())


channel.basic_consume(
    result.method.queue,
    callback,
    # 设置成 False，在调用callback函数时，未收到确认标识，消息会重回队列。True，无论调用callback成功与否，消息都被消费掉
    False)
channel.start_consuming()

模式二：direct

1. 传递或接受消息时需要指定 routing_key

这里发布了两种，10个OrderID和一个OrderIDd

# 发布
import json
import pika

credentials = pika.PlainCredentials('root', 'root')  # mq的用户名、密码
connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='10.131.70.49', port=5672,  # mq的IP、端口
    virtual_host='/',
    credentials=credentials
))
channel = connection.channel()
channel.exchange_declare(
    exchange='python-text',  # 指定队列
    durable=True,  # 消息持久化
    exchange_type='direct'  # 发布的模式
)
for i in range(10):
    message = json.dumps({"Order": "1000%s" % i})
    channel.basic_publish(
        exchange='python-text',
        routing_key='OrderID',
        body=message,
    )
message = json.dumps({"Order": "others"})
channel.basic_publish(
    exchange='python-text',
    routing_key='OrderIDd',
    body=message,
)
connection.close()

这里订阅OrderID会接收到十条消息

# 订阅OrderID
import pika

credentials = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='10.131.70.49', port=5672, virtual_host='/', credentials=credentials))
channel = connection.channel()
# 创建临时队列，consumer关闭后，队列自动删除
result = channel.queue_declare('', exclusive=True)
# 声明exchange，由exchange指定消息在哪个队列传递，如不存在，则创建。durable = True 代表exchange持久化存储，False 非持久化存储
channel.exchange_declare(exchange='python-text', durable=True, exchange_type='direct')
# 绑定exchange和队列  exchange 使我们能够确切地指定消息应该到哪个队列去
channel.queue_bind(
    exchange='python-text', queue=result.method.queue,routing_key='OrderID'
)


# 定义一个回调函数来处理消息队列中的消息，这里是打印出来
def callback(ch, method, properties, body):
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(body.decode())


channel.basic_consume(
    result.method.queue,
    callback,
    # 设置成 False，在调用callback函数时，未收到确认标识，消息会重回队列。True，无论调用callback成功与否，消息都被消费掉
    False)
channel.start_consuming()

这里订阅OrderIDd会接收到一条消息

# 订阅OrderIDd
import pika

credentials = pika.PlainCredentials('root', 'root')
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='10.131.70.49', port=5672, virtual_host='/', credentials=credentials))
channel = connection.channel()
# 创建临时队列，consumer关闭后，队列自动删除
result = channel.queue_declare('', exclusive=True)
# 声明exchange，由exchange指定消息在哪个队列传递，如不存在，则创建。durable = True 代表exchange持久化存储，False 非持久化存储
channel.exchange_declare(exchange='python-text', durable=True, exchange_type='direct')
# 绑定exchange和队列  exchange 使我们能够确切地指定消息应该到哪个队列去
channel.queue_bind(
    exchange='python-text', queue=result.method.queue,routing_key='OrderIDd'
)


# 定义一个回调函数来处理消息队列中的消息，这里是打印出来
def callback(ch, method, properties, body):
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(body.decode())


channel.basic_consume(
    result.method.queue,
    callback,
    # 设置成 False，在调用callback函数时，未收到确认标识，消息会重回队列。True，无论调用callback成功与否，消息都被消费掉
    False)
channel.start_consuming()