让 Celery 广播所有工作人员的返回结果

Question

有没有办法从 Celery Broadcast 任务中的每个工作人员那里获取所有结果？我想监控所有工人是否一切正常。任务被发送到的工作人员列表也将不胜感激。

Answer 1

不，这不容易。

但您不必将自己限制在内置的 amqp 结果后端， 您可以使用 Kombu (http://kombu.readthedocs.org) 发送自己的结果， 这是 Celery 使用的消息传递库：

from celery import Celery
from kombu import Exchange

results_exchange = Exchange('myres', type='fanout')

app = Celery()

@app.task(ignore_result=True)
def something():
    res = do_something()
    with app.producer_or_acquire(block=True) as producer:
        producer.send(
            {'result': res},
            exchange=results_exchange,
            serializer='json',
            declare=[results_exchange],
        )

producer_or_acquire 将使用 celery 创建一个新的kombu.Producer 代理连接池。