【问题标题】:Allow RabbitMQ and Pika maintain the conection always open允许 RabbitMQ 和 Pika 保持连接始终打开
【发布时间】:2019-05-27 08:47:53
【问题描述】:

我有一个从流中读取内容的 Python 脚本,当读取一个新字符串时,它会将其内容(一个字符串)推送到 RabbitMQ 队列。

问题是流可能不会在 1、2 或 9 小时左右发送消息,所以我希望 RabbitMQ 连接始终打开

问题是当我创建连接和通道时:

self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.host, credentials=self.credentials))
channel = self.connection.channel()
channel.exchange_declare(exchange=self.exchange_name, exchange_type='fanout')

...如果一个小时后有消息到达,我会收到此错误:

  File "/usr/local/lib/python3.7/asyncio/events.py", line 88, in _run
    self._context.run(self._callback, *self._args)
  File "/var/opt/rabbitmq-agent.py", line 34, in push_to_queue
    raise Exception("Error sending the message to the queue: " + format(e))
Exception: Error sending the message to the queue: Send message to publisher error: Channel allocation requires an open connection: <SelectConnection CLOSED socket=None params=<ConnectionParameters host=x port=xvirtual_host=/ ssl=False>>

我猜是rabbitmq服务器和客户端之间的连接已经关闭。

我怎样才能避免这种情况?我想要一个“请保持连接始终有效”。或许在 Pika 的连接参数中设置了一个超大的心跳?像这样的:

self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.host, credentials=self.credentials, heartbeat=6000))

任何其他更酷的解决方案都将受到高度赞赏。

提前致谢

【问题讨论】:

    标签: python rabbitmq pika rabbitmq-exchange


    【解决方案1】:

    我建议您每次在发送消息之前检查连接,如果连接已关闭,则只需重新连接。

    if not self.connection or self.connection.is_closed:
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.host, credentials=self.credentials))
        channel = self.connection.channel()
        channel.exchange_declare(exchange=self.exchange_name, exchange_type='fanout')
    

    【讨论】:

      【解决方案2】:

      您可以尝试将heartbeat 添加到您的ConnectionParameters。这将通过每隔指定秒发送一次心跳来创建轻流量。这将锻炼连接。一些防火墙或代理倾向于抓取空闲连接。甚至 RabbitMQ 在空闲的连接上也有超时。

      import pika
      
      # Set the connection parameters to connect to rabbit-server1 on port 5672
      # on the / virtual host using the username "guest" and password "guest"
      credentials = pika.PlainCredentials('guest', 'guest')
      parameters = pika.ConnectionParameters('rabbit-server1',
                                             5672,
                                             '/',
                                             heartbeat=60,
                                             credentials)
      

      有关 pika 文档,请参阅 here

      此外,您应该有适当的代码来缓解网络断开连接。这总是会发生并且将会发生。所以 appart 从心跳有一些异常处理准备以优雅的方式重新打开关闭的连接。

      【讨论】:

      • 这是错误的。 “heartbeat”参数只是告诉 RabbitMQ 服务器 RabbitMQ 服务器必须保持连接直到指定的值(以秒为单位),从客户端的最后一次请求时间开始。客户端必须定期发送心跳请求(周期应低于“心跳”选项中指定的时间)。 Pika 不会自动适应它。
      猜你喜欢
      • 2014-05-28
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2013-06-07
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2012-03-19
      相关资源
      最近更新 更多