python之使用 Kombu 设置 Rabbit MQ Heartbeat

编辑:

主要问题是第 3 方 rabbitmq 机器似乎时不时地终止空闲连接。那是我开始收到“Broken Pipe”异常的时候。获得通讯的唯一途径。恢复正常对我来说是终止进程并重新启动它们。我认为有更好的方法吗？

--

我有点迷路了。我正在连接到第 3 方 RabbitMQ 服务器以将消息推送到。时不时地，他们机器上的所有套接字都会掉线，我最终会收到“Broken Pipe”异常。

我被告知要在我的代码中实现心跳检查，但我不确定具体如何实现。我在这里找到了一些信息:http://kombu.readthedocs.org/en/latest/changelog.html#version-2-3-0但没有真正的示例代码。

我只需要在连接字符串中添加“?heartbeat=x”吗？ Kombu 会做剩下的事情吗？我看到我需要在“x/2”调用“Connection.heartbeat_check()”。我应该创建一个周期性任务来调用它吗？如何重新建立连接？

我正在使用:

• celery ==3.0.12
• 海带==2.5.4

我的代码现在看起来像这样。一个简单的 Celery 任务被调用以将消息发送到第 3 方 RabbitMQ 服务器(删除了日志记录和评论以使其简短，足够基本):

class SendMessageTask(Task): 
    name = "campaign.backends.send" 
    routing_key = "campaign.backends.send" 
    ignore_result = True 
    default_retry_delay = 60 # 1 minute. 
    max_retries = 5 
 
    def run(self, send_to, message, **kwargs): 
    payload = "Testing message" 
 
    try: 
        conn = BrokerConnection( 
        hostname=HOSTNAME, 
        port=PORT, 
        userid=USER_ID, 
        password=PASSWORD, 
        virtual_host=VHOST 
        ) 
 
        with producers[conn].acquire(block=True) as producer: 
        publish = conn.ensure(producer, producer.publish, errback=sending_errback, max_retries=3) 
        publish( 
            body=payload, 
            routing_key=OUT_ROUTING_KEY, 
            delivery_mode=2, 
            exchange=EXCHANGE, 
            serializer=None, 
            content_type='text/xml', 
            content_encoding = 'utf-8' 
        ) 
 
    except Exception, ex: 
        print ex 

感谢所有帮助。

请您参考如下方法:

虽然您当然可以为生产者添加心跳支持，但它对消费者进程更有意义。

启用心跳意味着你必须定期发送心跳，例如如果心跳设置为 1 秒，则必须每秒或更长时间发送一次心跳，否则远程将关闭连接。

这意味着您必须使用单独的线程或使用异步 io 来及时可靠地发送心跳，并且由于无法在线程之间共享连接，因此我们只能使用异步 io。

好消息是，将心跳添加到纯生产连接可能不会给您带来太多好处。