Celery 升级 (3.1->4.1) - 对等连接重置答案

【问题标题】：Celery upgrade (3.1->4.1) - Connection reset by peerCelery 升级 (3.1->4.1) - 对等连接重置
【发布时间】：2017-08-21 18:42:10
【问题描述】：

去年我们正在使用 celery，大约有 15 名工人，每个工人的并发定义在 1-4 之间。

最近我们将 celery 从 v3.1 升级到 v4.1

现在我们在每个工人日志中都有以下错误，任何想法可能导致此类错误？

2017-08-21 18:33:19,780 94794  ERROR   Control command error: error(104, 'Connection reset by peer') [file: pidbox.py, line: 46]
Traceback (most recent call last):
  File "/srv/dy/venv/lib/python2.7/site-packages/celery/worker/pidbox.py", line 42, in on_message
    self.node.handle_message(body, message)
  File "/srv/dy/venv/lib/python2.7/site-packages/kombu/pidbox.py", line 129, in handle_message
    return self.dispatch(**body)
  File "/srv/dy/venv/lib/python2.7/site-packages/kombu/pidbox.py", line 112, in dispatch
    ticket=ticket)
  File "/srv/dy/venv/lib/python2.7/site-packages/kombu/pidbox.py", line 135, in reply
    serializer=self.mailbox.serializer)
  File "/srv/dy/venv/lib/python2.7/site-packages/kombu/pidbox.py", line 265, in _publish_reply
    **opts
  File "/srv/dy/venv/lib/python2.7/site-packages/kombu/messaging.py", line 181, in publish
    exchange_name, declare,
  File "/srv/dy/venv/lib/python2.7/site-packages/kombu/messaging.py", line 203, in _publish
    mandatory=mandatory, immediate=immediate,
  File "/srv/dy/venv/lib/python2.7/site-packages/amqp/channel.py", line 1748, in _basic_publish
    (0, exchange, routing_key, mandatory, immediate), msg
  File "/srv/dy/venv/lib/python2.7/site-packages/amqp/abstract_channel.py", line 64, in send_method
    conn.frame_writer(1, self.channel_id, sig, args, content)
  File "/srv/dy/venv/lib/python2.7/site-packages/amqp/method_framing.py", line 178, in write_frame
    write(view[:offset])
  File "/srv/dy/venv/lib/python2.7/site-packages/amqp/transport.py", line 272, in write
    self._write(s)
  File "/usr/lib64/python2.7/socket.py", line 224, in meth
    return getattr(self._sock,name)(*args)
error: [Errno 104] Connection reset by peer

顺便说一句：我们的任务形式：

@app.task(name='EXAMPLE_TASK'],
          bind=True,
          base=ConnectionHolderTask)
def example_task(self, arg1, arg2, **kwargs):
    # task code

【问题讨论】：

标签： python rabbitmq celery amqp kombu

【解决方案1】：

我们在 celery 方面也遇到了很大的问题...我花了 20% 的时间与我们的工人一起解决奇怪的闲置挂起/崩溃问题叹息

我们有一个类似的案例，它是由高并发和高 worker_prefetch_multiplier 引起的，因为事实证明，获取数千个任务是破坏连接的好方法。

如果不是这样：尝试通过将 broker_pool_limit 设置为 None 来禁用代理池。

只是一些可能（希望）有所帮助的快速想法:-)

【讨论】：