Celery broadcast task not working

Question

I've tried to make a broadcast task but only one of my workers recieve it per each call. Would you please help me? (I'm using rabbitmq and node-celery)

default_exchange = Exchange('celery', type='direct')
celery.conf.update(
    CELERY_RESULT_BACKEND = "amqp",
    CELERY_RESULT_SERIALIZER='json',
    CELERY_QUEUES = (
        Queue('celery', default_exchange, routing_key='celery'),
        Broadcast('broadcast_tasks'),
    ),
    CELERY_ROUTES = (
        {'my_tasks.sample_broadcast_task': {
            'queue': 'broadcast_tasks',
            }},
        {'my_tasks.sample_normal_task': {
            'queue': 'celery',
            'exchange': 'celery',
            'exchange_type': 'direct',
            'routing_key': 'celery',
            }}
    ),
    )

I've also test following configurtion but not working.

celery.conf.update(
    CELERY_RESULT_BACKEND = "amqp",
    CELERY_RESULT_SERIALIZER='json',
    CELERY_QUEUES=(
        Queue('celery', Exchange('celery'), routing_key='celery'),
        Broadcast('broadcast'),
    ),
    )
@celery.task(ignore_result=True, queue='broadcast',
             options=dict(queue='broadcast'))
def sample_broadcast_task():
    print "test"

EDIT

after changing how to run worker by adding -Q broadcast, now i face to this error:

PreconditionFailed: Exchange.declare: (406) PRECONDITION_FAILED - inequivalent arg 'type' for exchange 'broadcast' in vhost '/': received 'direct' but current is 'fanout'

score 2 · Answer 1 · edited May 23 '17 at 11:46

After trying many many many things, i finally find a solution. This work for me. ( celery 3.1.24 (Cipater) and Python 2.7.12 )

WORKER - tasks.py :

from celery import Celery
import celery_config
from kombu.common import Broadcast, Queue, Exchange

app = Celery()
app.config_from_object(sysadmin_celery_config)

@app.task
def print_prout(x):
    print x
    return x

WORKER - celery_config.py :

# coding=utf-8
from kombu.common import Broadcast, Queue, Exchange


BROKER_URL = 'amqp://login:pass@172.17.0.1//'
CELERY_RESULT_BACKEND = 'redis://:login@172.17.0.1'


CELERY_TIMEZONE = 'Europe/Paris'
CELERY_ENABLE_UTC = True

CELERY_TASK_SERIALIZER = 'pickle'
CELERY_RESULT_SERIALIZER = 'pickle'
CELERY_ACCEPT_CONTENT = ['pickle', 'json', 'msgpack', 'yaml']
CELERY_DISABLE_RATE_LIMITS = True
CELERY_ALWAYS_EAGER = False

CELERY_QUEUES = (Broadcast('broadcast_tasks'), )

worker lauched with :

celery -A celery_worker.tasks worker --loglevel=info --concurrency=1 -n worker_name_1

On the client (another docker container for me).

from celery import Celery
from celery_worker import tasks

result = tasks.print_prout.apply_async(['prout'], queue='broadcast_tasks')

print result.get()

The next step for me is how to retrieve and display results returned by all the workers. The "print result.get()" seems to return only the result of the last worker. It does not seem obvious ( Have Celery broadcast return results from all workers )

score 1 · Answer 2 · answered Jan 05 '16 at 06:07

1

according to your description:

I've tried to make a broadcast task but only one of my workers recieve it per each call

you may be using direct type exchange.

answered Jan 05 '16 at 06:07

Chao Liu

21
2

score 0 · Answer 3 · answered Apr 02 '17 at 07:21

Try this

   from celery import Celery
   from kombu.common import Broadcast

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


   class CeleryConf:
   # List of modules to import when celery starts.
   CELERY_ACCEPT_CONTENT = ['json']
   CELERY_IMPORTS = ('main.tasks')
   CELERY_QUEUES = (Broadcast('q1'),)
   CELERY_ROUTES = {
    'tasks.sampletask': {'queue': 'q1'}
   }


   celeryapp = Celery('celeryapp', broker=BROKER_URL)
   celeryapp.config_from_object(CeleryConf())


   @celeryapp.task
   def sampletask(form):
      print form

To send the message, do

    d= sampletask.apply_async(['4c5b678350fc643'],serializer="json", queue='q1')

Celery broadcast task not working

3 Answers3