Celery on_worker_shutdown无法维持对象状态的原因及解决方法

1. 多进程内存隔离
   Celery默认采用prefork多进程模式运行Worker，主进程与子进程的内存空间完全独立。任务执行时，事件会被添加到子进程的EventsBuffer实例中，但worker_shutdown信号是在主进程触发的，主进程的缓冲区从未接收过任务事件，因此flush时显示0条数据。

2. 信号绑定的进程上下文问题
   你在EventsBuffer的__init__中绑定了worker_shutdown信号，但该绑定只对创建它的进程生效。子进程通过fork继承主进程的资源，但信号处理逻辑在fork后不会被子进程继承生效，导致子进程的缓冲区无法在自身关闭时触发flush。

3. 代码语法错误
   flush()方法缺少self参数，实际运行时会抛出TypeError，这是简化代码时的笔误，需修正为def flush(self):。

方案一：针对子进程绑定关闭信号（推荐单机器场景） #

改用worker_process_shutdown信号，该信号会在每个worker子进程关闭时触发，确保每个子进程的缓冲区都能被刷新。修改代码如下：

from celery import Task, Celery
from celery.signals import worker_process_shutdown  # 替换信号

class EventsBuffer:
    def __init__(self):
        self.buffer = []
        # 绑定子进程关闭信号
        worker_process_shutdown.connect(self._on_worker_shutdown)

    def add_event(self, event):
        self.buffer.append(event)
        if len(self.buffer) > 50:
            self.flush()

    def flush(self):  # 添加self参数
        print('Flushing the buffer with %s items' % len(self.buffer))
        # 执行实际的API写入逻辑

    def _on_worker_shutdown(self, **kwargs):
        print('Shutting down')
        self.flush()

class TaskBase(Task):
    def __init__(self):
        self.events_buffer = EventsBuffer()

app = Celery(__name__)

@app.task(base=TaskBase, bind=True)
def send_event(self, event):
    self.events_buffer.add_event(event)

启动Worker时保持默认prefork模式即可，每个子进程关闭时会自动刷新自己的缓冲区。

方案二：使用共享存储（推荐多机器/分布式场景） #

如果你的Worker是分布式部署的，进程内内存缓冲区无法跨进程/机器共享，此时应该用Redis、Memcached或数据库作为共享缓冲区。示例以Redis为例：

from celery import Task, Celery
from celery.signals import worker_shutdown
import redis

class EventsBuffer:
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379, db=0)
        self.key = 'events_buffer'
        worker_shutdown.connect(self._on_worker_shutdown)

    def add_event(self, event):
        self.redis.rpush(self.key, event)
        if self.redis.llen(self.key) > 50:
            self.flush()

    def flush(self):
        # 批量取出所有事件
        events = self.redis.lrange(self.key, 0, -1)
        print('Flushing the buffer with %s items' % len(events))
        # 执行API写入逻辑
        # 写入成功后清空缓冲区
        self.redis.delete(self.key)

    def _on_worker_shutdown(self, **kwargs):
        print('Shutting down')
        self.flush()

class TaskBase(Task):
    def __init__(self):
        self.events_buffer = EventsBuffer()

app = Celery(__name__)

@app.task(base=TaskBase, bind=True)
def send_event(self, event):
    self.events_buffer.add_event(event)

这种方式下，所有Worker进程共享同一个缓冲区，无论是定期flush还是Worker关闭时的flush，都能获取到所有未写入的事件。

方案三：单进程模式（仅用于测试） #

启动Worker时指定solo池，强制单进程运行，此时主进程与任务执行进程为同一个，缓冲区能共享：

celery -A your_app_name worker --pool=solo

该方案不适合生产环境，会导致Worker无法利用多核CPU，性能受限。

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