How can I set up Celery to call a custom worker initialization?

Question

I am quite new to Celery and I have been trying to setup a project with 2 separate queues (one to calculate and the other to execute). So far, so good.

My problem is that the workers in the execute queue need to instantiate a class with a unique object_id (one id per worker). I was wondering if I could write a custom worker initialization to initialize the object at start and keep it in memory until the worker is killed.

I found a similar question on custom_task but the proposed solution does not work in my case.

Considering the following toy example:

celery.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

initializing:

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

Is this the right approach? If so, what should I write in the configure_worker1 function in tasks.py to setup id at the worker initialization?

Thanks

The sender argument of worker_init is a WorkController instance, not a string, and in any case hostnames consists of a name and hostname combination, separated by @ (e.g. `name@example.com`). The celeryd_init signal can be used to connect by hostname: http://docs.celeryproject.org/en/latest/userguide/signals.html#celeryd-init — asksol, Nov 03 '14 at 21:54
@asksol sorry, my mistake when pasting the code. Indeed I have name@hostname... I will edit the question. I just can't initialize an object in the worker. Something like self.x = 1... — b3rt0, Nov 04 '14 at 13:19

score 8 · Accepted Answer · answered Nov 06 '14 at 19:58

I found out the answer by following this http://docs.celeryproject.org/en/latest/userguide/tasks.html#instantiation

The tasks.py looks like this:

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

initializing as before:

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

How to pass arguments to the init function of Task1 and Task2 — maple, Oct 31 '18 at 05:41

How can I set up Celery to call a custom worker initialization?

1 Answers1

Linked