In my previous blog post we looked into custom queues and task routing. We had to configure per task which queue we want Celery to task a route to. While this approach works well for a simple setup, it does not scale well for applications and micro-services where many Celery tasks need to be routed to a number of different worker queues.

Step 1: Celery task_routes config

Instead of configuring the task_routes per task, which queue you want to route a task to, you can tell Celery to use a custom class instead by specifying the path to said class (also, have a look at the Celery docs available at http://docs.celeryproject.org/en/latest/userguide/routing.html):

app = Celery(__name__)
app.conf.update({
    'broker_url': os.environ['CELERY_BROKER_URL'],
    'imports': (
        'tasks',
    ),
    'task_routes': ('task_router.TaskRouter',),
    'task_serializer': 'json',
    'result_serializer': 'json',
    'accept_content': ['json']})</code></pre>

Step 2: Define TaskRouter class

As per our task_routes value above, we need to define the custom TaskRouter
class in the module task_router.py. Celery expects the method route_for_task
that passes the task name as its first argument. Note how the method returns a dict that looks
exactly like the one used for manual task routing.

class TaskRouter:
    def route_for_task(self, task, *args, **kwargs):
        if ':' not in task:
            return {'queue': 'default'}

        namespace, _ = task.split(':')
        return {'queue': namespace}

Our idea is to route a task based on its task name, in particular we want to assume that our task
names follow the pattern queue:taskname. In our previous blog post's example
we had a task named fetch_bitcoin_price_index that we wanted to be routed to a queue
called feeds. We rename this task to feeds:fetch_bitcoin_price_index.

@app.task(bind=True, name='feeds:fetch_bitcoin_price_index')
def fetch_bitcoin_price_index(self, start_date, end_date):
   ...


@app.task(bind=True, name='filters:calculate_moving_average')
def calculate_moving_average(self, args, window):
    ...

We need to run two Celery workers. One subscribes to the feeds,
the other one to the filters queue:

~$ celery worker --app=worker.app --hostname=worker.feeds@%h --queues=feeds
~$ celery worker --app=worker.app --hostname=worker.filters@%h --queues=filters

Note the --queues command line arguments. They your workers subscribe to particular queues.
For subscribing to more than one queue, use a comma-separated list, like so --queues=feeds,filters.
For further information, have a look at the Celery docs.

Step 3: Ready for action

Bring up the docker-compose stack and run example.py:

# start up stack
~$ docker-compose up -d

# execute python example.py in container
~$ docker-compose exec worker-feeds python example.py --start_date=2018-01-01 --end_date=2018-05-29 --window=3

The script invokes the Celery chain that consists of two tasks: fetch_bitcoin_price_index fetches
Bicoin Price Index data from the Coindesk API via the feeds queue to the worker-feeds Celery worker. When the task completes successfully, the result is passed onto the calculate_moving_average via the filters queue to the worker-filters Celery worker.

Check the docker-compose logs to follow the task flow through the two workers:

~$ docker-compose logs -f

The docker-compose.yml stack also comes with a flower instance.
Flower a tool for monitoring Celery workers and tasks. Check out your browser on http://localhost:5555.

Summary

In this blog post you learned how to configure Celery to route tasks using a custom task router. This solution scales well when using many tasks across many queues and workers.