之前只是用celery, 这次用一下pika

参考rabbitMQ官网的python版,https://www.rabbitmq.com/tutorials/tutorial-one-python.html

没想到各种坑.

如果说rabbitMQ官网是为了让新人入门,所以刻意忽略掉细节, 那么必须吐槽pika的官方文档, 很不好.远不如celery

1 Stream connection lost: BrokenPipeError(32, ‘Broken pipe‘)

使用pika 的BlockingConnection

raise self._closed_result.value.error
pika.exceptions.StreamLostError: Stream connection lost: BrokenPipeError(32, ‘Broken pipe‘)

根据https://www.cnblogs.com/zhaof/p/9774390.html

是要在连接时设置心跳为0,就不会超时自动下线了, 否则RabbitMQ服务器会发过来默认值580

#--------------rabbitMQ------------------
import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(
    host=‘localhost‘,
    heartbeat=0, #never exit after start
    ))

channel = connection.channel()


channel.queue_declare(queue=‘update_sql‘)

这个错误在测试消费端时没测出来,因为测试使用的发布者和官方文档里一样,发完就下线退出了. 这样当然看不出这个心跳问题.

但是联调时就暴露了. 真无语.

2  content_type

默认的body是二进制的. 然后消费端要

    channel.basic_publish(‘exchange_name‘,
                          ‘routing_key‘,
                          ‘Test Message‘,
                          pika.BasicProperties(content_type=‘text/plain‘,
                                               type=‘example‘))

        self.channel.basic_publish(exchange=exchange, routing_key=routing_key, body=message,
properties=pika.BasicProperties(delivery_mode=2,message_id=message_id,content_type="application/json"))

3 ack和durable

ack防止消费者出问题, durable防止rabbitMQ服务器本身出问题

所以ack在消费端定义

    channel.basic_consume(queue=‘update_sql‘,
                        auto_ack=False,
                        on_message_callback=callback)

而durable在channel里队列声明里 在 生产端,消费端都要统一声明队列

channel.queue_declare(queue=‘update_sql‘, durable=True, exclusive=False, auto_delete=False)

引用 https://blog.csdn.net/hlxx55/article/details/80964440

ack

rabbitMQ是默认开启自动应答的，这样当rabbitMQ将消息发给消费者，就会从内存中将消息删除，这样会带来一个问题，如果消费者未处理完消息而宕机，那么消息就会丢失。所以，我们将自动应答关闭，当rabbitMQ收到消费者处理完消息的回应后才会从内存中删除消息。

 

durable

rabbitMQ默认将消息存储在内存中，若rabbitMQ宕机，那么所有数据就会丢失，所以在声明队列的时候可以声明将数据持久化，但是如果已经声明了一个未持久化的队列，那么不能修改，只能将这个队列删除或重新声明一个持久化数据。

pika 与 rabbitMQ 阻塞连接

原文：https://www.cnblogs.com/xuanmanstein/p/11561277.html