Celery搭配gevent Worker出现BrokenPipeError的解决方法咨询

问题背景 #

我们的应用采用Celery搭配gevent worker，worker以独立Docker容器运行，启动命令：
celery -A app.worker worker --pool=gevent --loglevel=info --concurrency=50 -E -n woker-1 -Q some_q

每日会出现一次（有时更频繁）BrokenPipeError，报错栈如下：

2023-04-14 05:06:11,023 ERROR celery.worker.consumer.consumer perform_pending_operations() L227  Pending callback raised: BrokenPipeError(32, 'Broken pipe')
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/celery/worker/consumer/consumer.py", line 225, 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

错误根源在/usr/local/lib/python3.8/site-packages/gevent/_socketcommon.py，已关联Celery的两个相关Issue：旧版本的#3377和新版本未解决的#7888。

错误发生后，容器未停止，但worker停止工作，与RabbitMQ的连接中断，无法处理新任务。

现有临时方案 #

通过脚本定时重启容器，相关配置如下：

重启脚本 #

# some pseudo python app code
declare -i n=0

while [ $n -lt 15 ]
do
    d=$(date)
    n=n+1
    echo $d Step: $n
    sleep 2
done
# some sleeping after it, then exit

exit 0  # the most important is to exit after n minutes of sleeping

Docker Compose #

version: '3.8'
services:
  xyz:
    image: xyz:latest
    container_name: xyz
    hostname: xyz
    restart: always  # always restart app despite no errors - code 0

Dockerfile #

FROM python:3.8

RUN mkdir -p /my_super_app
COPY ./app /my_super_app/app

WORKDIR /my_super_app

# celery-worker starts inside the bash script
CMD ["bash", "/bgate-smev/app/worker-start.sh"] 

稳健解决方案建议 #

1. 配置Celery连接自动重连 #

在Celery配置中启用连接自动重连，并调整相关参数：

# app/worker.py 中的Celery配置
CELERY_BROKER_HEARTBEAT = 30
CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = True
CELERY_BROKER_CONNECTION_MAX_RETRIES = 10
CELERY_BROKER_CONNECTION_RETRY = True

这些配置会让Celery在连接断开时自动尝试重连，避免worker直接挂死。

2. 替换Gevent Pool为Eventlet或Prefork #

如果Gevent的socket兼容性问题无法解决，可以尝试切换到Eventlet pool，启动命令改为：
celery -A app.worker worker --pool=eventlet --loglevel=info --concurrency=50 -E -n woker-1 -Q some_q
或者使用默认的prefork池（适合CPU密集型任务，IO密集型也可尝试），调整concurrency参数适配容器资源。

3. 捕获并处理BrokenPipeError #

通过Celery的信号机制，在worker层面捕获异常并触发重连：

from celery.signals import worker_process_init
import gevent.signal
import signal

def handle_broken_pipe(signalnum, frame):
    # 触发worker重连逻辑
    from celery import current_app
    current_app.connection().reconnect()

@worker_process_init.connect
def setup_signal_handlers(sender=None, **kwargs):
    gevent.signal(signal.SIGPIPE, handle_broken_pipe)

注意：此方法需要确保信号处理逻辑不会引入新的线程/协程冲突。

4. 升级依赖版本 #

尝试升级Celery、Kombu、Gevent到最新稳定版，部分旧版本的兼容性问题可能已被修复：

pip install --upgrade celery kombu gevent

5. 监控worker状态并自动重启（比定时重启更智能） #

使用Docker健康检查监控worker状态，当检测到worker无法连接RabbitMQ或停止处理任务时自动重启容器：

version: '3.8'
services:
  xyz:
    image: xyz:latest
    container_name: xyz
    hostname: xyz
    restart: on-failure
    healthcheck:
      test: ["CMD", "celery", "-A", "app.worker", "inspect", "ping", "-d", "woker-1"]
      interval: 60s
      timeout: 10s
      retries: 3
      start_period: 30s

当健康检查失败时，Docker会自动重启容器，比定时重启更精准。

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