Gevent模式下Celery Worker与RabbitMQ连接出现BrokenPipeError求助

生产环境用Gevent作为Celery Worker并发池时，频繁触发与RabbitMQ连接的BrokenPipeError，开发环境因负载低未复现，就算提升服务器CPU资源（最高使用率仅50%），问题依然存在。

错误栈信息 #

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/celery/worker/consumer/consumer.py", line 246, in perform_pending_operations
    self._pending_operations.pop()()
  File "/usr/local/lib/python3.8/site-packages/vine/promises.py", line 160, in __call__
    return self.throw()
  File "/usr/local/lib/python3.8/site-packages/vine/promises.py", line 157, in __call__
    retval = fun(*final_args, **final_kwargs)
  File "/usr/local/lib/python3.8/site-packages/kombu/message.py", line 128, in ack_log_error
    self.ack(multiple=multiple)
  File "/usr/local/lib/python3.8/site-packages/kombu/message.py", line 123, in ack
    self.channel.basic_ack(self.delivery_tag, multiple=multiple)
  File "/usr/local/lib/python3.8/site-packages/amqp/channel.py", line 1407, in basic_ack
    return self.send_method(
  File "/usr/local/lib/python3.8/site-packages/amqp/abstract_channel.py", line 70, in send_method
    conn.frame_writer(1, self.channel_id, sig, args, content)
  File "/usr/local/lib/python3.8/site-packages/amqp/method_framing.py", line 186, in write_frame
    write(buffer_store.view[:offset])
  File "/usr/local/lib/python3.8/site-packages/amqp/transport.py", line 347, in write
    self._write(s)
  File "/usr/local/lib/python3.8/site-packages/gevent/_socketcommon.py", line 699, in sendall
    return _sendall(self, data_memory, flags)
  File "/usr/local/lib/python3.8/site-packages/gevent/_socketcommon.py", line 409, in _sendall
    timeleft = __send_chunk(socket, chunk, flags, timeleft, end)
  File "/usr/local/lib/python3.8/site-packages/gevent/_socketcommon.py", line 338, in __send_chunk
    data_sent += socket.send(chunk, flags)
  File "/usr/local/lib/python3.8/site-packages/gevent/_socketcommon.py", line 722, in send
    return self._sock.send(data, flags)
BrokenPipeError: [Errno 32] Broken pipe

当前配置 #

Celery启动命令

celery -A analytics worker -P gevent -c 500 -l info -E --without-gossip --without-mingle --without-heartbeat &

Django中的Celery配置

CELERY_IGNORE_RESULT = True
CELERY_WORKER_PREFETCH_MULTIPLIER = 100
CELERY_WORKER_MAX_TASKS_PER_CHILD = 400
CELERYD_TASK_SOFT_TIME_LIMIT = 60 * 60 * 12
CELERYD_TASK_TIME_LIMIT = 60 * 60 * 13

问题分析 #

这个错误是Worker在给RabbitMQ发送任务ACK时，连接已经被断开导致的。结合生产高负载场景，主要诱因有这几点：

1. Gevent协程数设置过高，导致RabbitMQ连接资源耗尽，或者单连接并发过载
2. 任务预取倍数太大，Worker一次性拉取大量任务后，连接长时间闲置被RabbitMQ主动断开
3. 禁用了Worker心跳，RabbitMQ无法检测连接存活状态，超时后直接切断连接

解决方案 #

1. 降低Gevent协程数

当前-c 500的协程数远超合理范围（Gevent协程数建议100-200之间，根据服务器内存调整），过多协程会导致TCP连接频繁创建销毁，或者单连接请求过载。修改启动命令：

celery -A analytics worker -P gevent -c 150 -l info -E --without-gossip --without-mingle &

2. 调低任务预取乘数

CELERY_WORKER_PREFETCH_MULTIPLIER = 100会让Worker一次性拉取100倍于协程数的任务，高负载下连接长时间无交互，被RabbitMQ心跳机制断开。建议调整到1-10之间：

CELERY_WORKER_PREFETCH_MULTIPLIER = 5

3. 恢复Worker心跳机制

启动命令中--without-heartbeat禁用了心跳，导致RabbitMQ无法感知连接状态，超时后直接断开。移除该参数，让Celery维持正常心跳：

celery -A analytics worker -P gevent -c 150 -l info -E --without-gossip --without-mingle &

4. 优化RabbitMQ连接配置

在Django配置中增加连接稳定性相关参数：

BROKER_HEARTBEAT = 30
BROKER_CONNECTION_TIMEOUT = 30
BROKER_CONNECTION_RETRY = True
BROKER_CONNECTION_MAX_RETRIES = 10

5. 调整子进程任务上限

当前CELERY_WORKER_MAX_TASKS_PER_CHILD = 400可适当调低，让Worker进程更频繁重启，避免长时间运行导致的连接资源泄漏：

CELERY_WORKER_MAX_TASKS_PER_CHILD = 200

内容的提问来源于stack exchange，提问作者rmaleki