Enrutar la tarea de apio a la cola específica

Question

Tengo dos procesos apicales separados que se ejecutan en mi servidor, administrados por supervisor. Se establecen para escuchar en colas separadas como tal:

[program:celeryd1] 
command=/path/to/celeryd --pool=solo --queues=queue1 
... 

[program:celeryd2] 
command=/path/to/celeryd --pool=solo --queues=queue2 
... 

Y mi celeryconfig es como la siguiente:

from celery.schedules import crontab 

BROKER_URL = "amqp://guest:guest@localhost:5672//" 

CELERY_DISABLE_RATE_LIMITS = True 
CELERYD_CONCURRENCY = 1 
CELERY_IGNORE_RESULT = True 

CELERY_DEFAULT_QUEUE = 'default' 
CELERY_QUEUES = { 
    'default': { 
     "exchange": "default", 
     "binding_key": "default", 
    }, 
    'queue1': { 
     'exchange': 'queue1', 
     'routing_key': 'queue1', 
    }, 
    'queue2': { 
     'exchange': 'queue2', 
     'routing_key': 'queue2', 
    }, 
} 

CELERY_IMPORTS = ('tasks',) 

CELERYBEAT_SCHEDULE = { 
    'first-queue': { 
     'task': 'tasks.sync', 
     'schedule': crontab(hour=02, minute=00), 
     'kwargs': {'client': 'client_1'}, 
     'options': {'queue': 'queue1'}, 
    }, 
    'second-queue': { 
     'task': 'tasks.sync', 
     'schedule': crontab(hour=02, minute=00), 
     'kwargs': {'client': 'client_2'}, 
     'options': {'queue': 'queue1'}, 
    }, 
} 

Todos tasks.sync tareas deben ser enviados a una cola específica (y el progreso, por lo tanto celeryd). Pero cuando trato de ejecutar la tarea manualmente con sync.apply_async(kwargs={'client': 'value'}, queue='queue1'), ambos empleados aprenden a realizar la tarea. ¿Cómo puedo hacer que la ruta de la tarea sea la correcta y solo la ejecute el trabajador vinculado a la cola?

Answer 1

Usted solo está ejecutando una instancia de celerybeat ¿verdad?

¿Quizás tengas enlaces de cola antiguos que chocan con esto? Intente ejecutar rabbitmqctl list_queues y rabbitmqctl list_bindings, , tal vez restablezca los datos en el intermediario para comenzar desde cero.

El ejemplo que tienes aquí debería funcionar, y está funcionando para mí cuando lo probé.

Sugerencia: Dado que está utilizando el mismo valor de exchange y binding_key que el nombre de la cola, , no es necesario que los liste explícitamente en CELERY_QUEUES. Cuando CELERY_CREATE_MISSING_QUEUES está activado (que es lo predeterminado), las colas se crearán automáticamente exactamente igual que tiene si solo hace celeryd -Q queue1 o envía una tarea a una cola indefinida.