Pika心跳终止连接

Question

我有一个代码，只是用 pika 将消息排队到代理队列中。

class Publisher:

    def __init__(self, config):
        self._params = ConnectionParameters(
            host = config.RABBITMQ_HOST,
            credentials = PlainCredentials(config.RABBITMQ_USER, config.RABBITMQ_PASSWORD))
        self._conn = None
        self._channel = None
        self.exchange_name = config.RABBITMQ_AGENT_EXCHANGE


    def connect(self):
        if not self._conn or self._conn.is_closed:
            self._conn = BlockingConnection(self._params)
            self._channel = self._conn.channel()
            self._channel.exchange_declare(exchange=self.exchange_name,  exchange_type = 'topic')

    def _publish(self, task):
        properties = BasicProperties(expiration=task.expiration_ms)
        self._channel.basic_publish(exchange= self.exchange_name,
                                    routing_key = task.routing_key,
                                    properties = properties if task.has_expiration else None,
                                    body=dumps(task, cls = TaskEncoder).encode())
        logging.debug('message sent: %s', task)


    def publish(self, msg):
        """Publish msg, reconnecting if necessary."""

        try:
            self._publish(msg)
        except ConnectionClosed:
            logging.error('reconnecting to queue')
            self.connect()
            self._publish(msg)

Pika 停止将消息排队以与下一条消息进行长时间连接，并且不再抛出任何错误

2021-03-14 12:25:09,981 MainThread-140100212655936 pika.heartbeat [INFO] - Connection is idle, 1 stale byte intervals
2021-03-14 12:25:09,981 MainThread-140100212655936 pika.adapters.utils.io_services_utils [INFO] - Aborting transport connection: state=1; <socket.socket fd=6, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('192.168.184.108', 41024), raddr=('10.100.176.158', 5672)>
2021-03-14 12:25:09,981 MainThread-140100212655936 pika.adapters.utils.io_services_utils [INFO] - _AsyncTransportBase._initate_abort(): Initiating abrupt asynchronous transport shutdown: state=1; error=None; <socket.socket fd=6, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('192.168.184.108', 41024), raddr=('10.100.176.158', 5672)

代码使用

publisher = Publisher(config)
publisher.connect()
while True:
  publisher.publish(obj)
  time.sleep(1)

我有两个问题：

如何预防呢？这种情况下禁用心跳可以吗？

如何使用防火墙重现/模拟此行为？我尝试添加一条在 RMQ 端口上丢包的规则，但没有成功。

皮卡版本：1.0.1

RMQ版本：3.8.9

蟒蛇：3.8.6

Answer 1

有相当多的帖子建议保持回调执行时间非常短，因此控制权应该返回到 pika。 在这里查看更多信息

如果你有大的事情要计算，最好将它们跨越到另一个线程/进程，或者累积它们以供以后执行。这个概念对我使用其他实时 API 很有帮助。

似乎发生的事情是，兔子正在关闭通道（如果你的手不忙于其他地方，你可能会在回调中发现这一点），并且在我看来，鼠兔的心跳似乎停止了。

• 链接