Python与RabbitMQ

缺乏、安全感 2023-06-12 08:48 7阅读 0赞

# Demo #

消费者：

# 消费者
    import pika
    
    # 连接
    # auth info
    auth = pika.PlainCredentials('guest', 'guest')  # auth info
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        '127.0.0.1', 5672, '/', auth))  # connect to rabbit
    channel = connection.channel()  # create channel
    
    # 申明队列
    channel.queue_declare(queue='hello', durable=True, passive=True)
    
    
    def callback(ch, method, properties, body):
    # ack
        ch.basic_ack(delivery_tag=method.delivery_tag)
        print(body.decode())
    
    
    # no_ack 设置成 False，在调用callback函数时，未收到确认标识，消息会重回队列
    # True，无论调用callback成功与否，消息都被消费掉
    channel.basic_consume("hello", callback,
                          False)
    channel.start_consuming()

生产者：

# 生产者
    import pika
    
    # 连接
    # auth info
    auth = pika.PlainCredentials('guest', 'guest')
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        '127.0.0.1', 5672, '/', auth))  # connect to rabbit
    channel = connection.channel()
    
    """
    MQ默认建立的是临时 queue 和 exchange，如果不声明持久化，一旦 rabbitmq 挂掉，queue、exchange 将会全部丢失
    所以一般在创建 queue 或者 exchange 的时候会声明 持久化
    """
    # 声明消息队列，消息将在这个队列传递，如不存在，则创建,durable = True为持久化
    channel.queue_declare(queue='hello')
    """
    虽然 exchange 和 queue 都申明了持久化，但如果消息只存在内存里rabbitmq 重启后
    内存里的东西还是会丢失。所以必须声明消息也是持久化，从内存转存到硬盘。
     properties=pika.BasicProperties(delivery_mode = 2))
    """
    # 发送消息
    channel.basic_publish(exchange='',
                          routing_key='hello',
                          body='Flash22',
                          properties=pika.BasicProperties(delivery_mode=2))  # body是msg内容
    connection.close()

# 消息持久化 #

一条消息进行持久化具体可以分为4步：

*  queue持久化
 *  exchange 持久化
 *  消息持久化
 *  ack机制

代码分别如下：

### queue持久化 ###

# durable = True 代表消息队列持久化存储，False 非持久化存储
    result = channel.queue_declare(queue = 'hello',durable = True)

### exchange 持久化 ###

# durable = True 代表exchange持久化存储，False 非持久化存储
    channel.exchange_declare(exchange = 'hello', durable = True)

**注意：如果已存在一个非持久化的 queue 或 exchange ，执行上述代码会报错，因为当前状态不能更改 queue 或 exchange 存储属性，需要删除重建。如果 queue 和 exchange 中一个声明了持久化，另一个没有声明持久化，则不允许绑定。 **

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70][]

### 消息持久化 ###

# 向队列插入数值 routing_key是队列名。delivery_mode = 2 声明消息在队列中持久化，delivery_mod = 1 消息非持久化
        channel.basic_publish(exchange = '',routing_key = 'hello',body = message,
                              properties=pika.BasicProperties(delivery_mode = 2))

### ack机制 ###

**    消费者（consumer）调用callback函数时，会存在处理消息失败的风险，如果处理失败，则消息丢失。但是也可以选择消费者处理失败时，将消息回退给 rabbitmq ，重新再被消费者消费，这个时候需要设置确认标识。**

# 回调函数
    def callback(ch, method, properties, body):
    #ack确认
        ch.basic_ack(delivery_tag=method.delivery_tag)
        print(body.decode())
    
    
    # no_ack 设置成 False，在调用callback函数时，未收到确认标识，消息会重回队列
    # True，无论调用callback成功与否，消息都被消费掉
    channel.basic_consume("hello", callback,
                          auto_ack=False)

RabbitMQ基础相关：[https://blog.csdn.net/qq\_37598011/article/details/87869792][https_blog.csdn.net_qq_37598011_article_details_87869792]

# **fanout、direct具体实现** #

**fanout：它会把所有发送到该交换器的消息路由到所有与该交换器绑定的队列中。**

**消费者：**

import pika
    
    # 连接
    # auth info
    auth = pika.PlainCredentials('guest', 'guest')  # auth info
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        '127.0.0.1', 5672, '/', auth))  # connect to rabbit
    channel = connection.channel()  # create channel
    
    # 创建临时队列，consumer关闭后，队列自动删除
    result = channel.queue_declare('', exclusive=True)
    # 声明exchange，由exchange指定消息在哪个队列传递，如不存在，则创建。durable = True 代表exchange持久化存储，False 非持久化存储
    channel.exchange_declare(exchange='hello_fanout', durable=True, exchange_type='fanout')
    # 绑定exchange和队列  exchange 使我们能够确切地指定消息应该到哪个队列去
    channel.queue_bind(exchange='hello_fanout', 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()

**生产者：**

import json
    
    import pika
    
    # auth info
    auth = pika.PlainCredentials('guest', 'guest')
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        '127.0.0.1', 5672, '/', auth))  # connect to rabbit
    channel = connection.channel()
    
    # 声明exchange，由exchange指定消息在哪个队列传递，如不存在，则创建。durable = True 代表exchange持久化存储，False 非持久化存储
    channel.exchange_declare(exchange='hello_fanout', durable=True, exchange_type='fanout')
    for i in range(10):
        message = json.dumps({'OrderId': i})
        # 向队列插入数值 routing_key是队列名。delivery_mode = 2 声明消息在队列中持久化，delivery_mod = 1 消息非持久化。routing_key 不需要配置
        channel.basic_publish(exchange='hello_fanout', routing_key='', body=message,
                              properties=pika.BasicProperties(delivery_mode=2))
        print(message)
    connection.close()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 1][]

**direct**：direct 类型的交换器它会把消息路由到那些 BindingKey 和 RoutingKey 完全匹配的队列中。

**生产者：**

import json
    
    import pika
    
    # auth info
    auth = pika.PlainCredentials('guest', 'guest')
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        '127.0.0.1', 5672, '/', auth))  # connect to rabbit
    channel = connection.channel()
    
    # 声明exchange，由exchange指定消息在哪个队列传递，如不存在，则创建
    # durable = True 代表exchange持久化存储，False 非持久化存储
    channel.exchange_declare(exchange='direct_producer', durable=True, exchange_type='direct')
    
    for i in range(10):
        message = json.dumps({'ProducerId': i})
        # 指定 routing_key。delivery_mode = 2 声明消息在队列中持久化，delivery_mod = 1 消息非持久化
        channel.basic_publish(exchange='direct_producer', routing_key='routingKey1', body=message,
                              properties=pika.BasicProperties(delivery_mode=2))
        print(message)
    connection.close()

**消费者：**

import pika
    
    # 连接
    # auth info
    auth = pika.PlainCredentials('guest', 'guest')  # auth info
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        '127.0.0.1', 5672, '/', auth))  # connect to rabbit
    channel = connection.channel()  # create channel
    
    # 创建临时队列，consumer关闭后，队列自动删除
    result = channel.queue_declare('', exclusive=True)
    # 声明exchange，由exchange指定消息在哪个队列传递，如不存在，则创建。durable = True 代表exchange持久化存储，False 非持久化存储
    channel.exchange_declare(exchange='direct_producer', durable=True, exchange_type='direct')
    # 绑定exchange和队列  exchange 使我们能够确切地指定消息应该到哪个队列去
    channel.queue_bind(exchange='direct_producer', queue=result.method.queue, routing_key='routingKey1')
    
    
    # 定义一个回调函数来处理消息队列中的消息，这里是打印出来
    def callback(ch, method, properties, body):
        ch.basic_ack(delivery_tag=method.delivery_tag)
        print(body.decode())
    
    
    # channel.basic_qos(prefetch_count=1)
    channel.basic_consume(result.method.queue, callback,
                          # 设置成 False，在调用callback函数时，未收到确认标识，消息会重回队列。True，无论调用callback成功与否，消息都被消费掉
                          False)
    channel.start_consuming()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 2][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191111144426396.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_qq_37598011_article_details_87869792]: https://blog.csdn.net/qq_37598011/article/details/87869792
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191111150240702.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20191111150906137.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx,size_16,color_FFFFFF,t_70