python - 使用 Kombu 设置 Rabbit MQ Heartbeat

Question

编辑:

主要问题是第 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

感谢所有帮助。

Answer 1

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

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

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

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