如何设置 Celery 来调用自定义工作器初始化？答案

【问题标题】：How can I set up Celery to call a custom worker initialization?如何设置 Celery 来调用自定义工作器初始化？
【发布时间】：2014-12-29 12:14:57
【问题描述】：

我对 Celery 很陌生，我一直在尝试设置一个具有 2 个独立队列的项目（一个用于计算，另一个用于执行）。到目前为止，一切都很好。

我的问题是执行队列中的工作人员需要实例化一个具有唯一 object_id 的类（每个工作人员一个 id）。我想知道是否可以编写自定义工作程序初始化以在启动时初始化对象并将其保存在内存中，直到工作人员被杀死。

我在custom_task 上发现了一个类似的问题，但建议的解决方案不适用于我的情况。

考虑以下玩具示例：

芹菜.py

from celery import Celery

app = Celery('proj',
             broker='amqp://guest@localhost//',
             backend='amqp://',
             include=['proj.tasks'])

app.conf.update(
    CELERY_TASK_RESULT_EXPIRES=60,
    CELERY_ROUTES = {"proj.tasks.add1": {"queue": "q1"}},
)

if __name__ == '__main__':
    app.start()

tasks.py

from proj.celery import app
from celery.signals import worker_init

@worker_init.connect(sender='worker1@hostname')
def configure_worker1(*args, **kwargs):
    #SETUP id=1 for add1 here???

@worker_init.connect(sender='worker2@hostname')
def configure_worker2(*args, **kwargs):
    #SETUP id=2 for add1 here???

@app.task
def add1(y):
    return id + y

@app.task
def add(x, y):
    return x + y

初始化：

celery multi start worker1 -A proj -l info -Q q1
celery multi start worker2 -A proj -l info -Q q1
celery multi start worker3 -A proj -l info

这是正确的方法吗？如果是这样，我应该在tasks.py中的configure_worker1函数中写什么来在worker初始化时设置id？

谢谢

【问题讨论】：

worker_init 的 sender 参数是一个 WorkController 实例，而不是字符串，并且在任何情况下，主机名都由名称和主机名组合组成，以 @ 分隔（例如 name@example.com）。 celeryd_init 信号可用于通过主机名连接：docs.celeryproject.org/en/latest/userguide/…
@asksol 抱歉，我在粘贴代码时出错了。确实我有 name@hostname... 我会编辑这个问题。我只是无法在工作人员中初始化对象。像 self.x = 1...

标签： python celery

【解决方案1】：

我通过关注http://docs.celeryproject.org/en/latest/userguide/tasks.html#instantiation找到了答案

tasks.py 如下所示：

from proj.celery import app
from celery import Task

class Task1(Task):
    def __init__(self):
        self._x = 1.0

class Task2(Task):
    def __init__(self):
        self._x = 2.0

@app.task(base=Task1)
def add1(y):
    return add1._x + y

@app.task(base=Task2)
def add2(y):
    return add2._x + y

像以前一样初始化：

celery multi start worker1 -A proj -l info -Q q1
celery multi start worker2 -A proj -l info -Q q1
celery multi start worker3 -A proj -l info

【讨论】：

如何将参数传递给Task1和Task2的init函数