celery-demo

client.py

#!/usr/bin/env python
# encoding: utf-8
from worker import add

# add.apply_async((1, ), priority=1)
# add.apply_async((1, ), priority=1)
add.apply_async((7, ), priority=7)
add.apply_async((6, ), priority=6)
add.apply_async((5, ), priority=5)
add.apply_async((8, ), priority=8)
add.apply_async((9, ), priority=9)

route_client.py

#!/usr/bin/env python
# encoding: utf-8
from route_worker import add

# add.apply_async((1, ), priority=1)
# add.apply_async((1, ), priority=1)
add.apply_async((7, ), queue="for_task_A", priority=7)
add.apply_async((7, ), queue="for_task_B", priority=7)
# add.apply_async((7, ), routing_key='for_task_A', priority=7)

route_worker.py

#!/usr/bin/env python
# encoding: utf-8
import time

from kombu import Queue, Exchange
from celery import Celery

app = Celery('worker', broker='redis://localhost:6379/1')
app.conf.update({
    "CELERY_QUEUES": (
        Queue('default', Exchange('default'), routing_key='default'),
        Queue('for_task_A', Exchange('for_task_A'), routing_key='for_task_A'),
        Queue('for_task_B', Exchange('for_task_B'), routing_key='for_task_B'),
    ),
    "CELERY_ROUTES": {
        'route_worker.add': {'queue': 'for_task_A', 'routing_key': 'for_task_A'},
        'route_worker.sub': {'queue': 'for_task_B', 'routing_key': 'for_task_B'},
    }
})

@app.task
def add(pri):
    print "add called with pri: {}".format(pri)
    time.sleep(5)
    return pri

@app.task
def sub(pri):
    print "sub called with pri: {}".format(pri)
    time.sleep(5)
    return pri

worker.py

#!/usr/bin/env python
# encoding: utf-8
import time

from celery import Celery

app = Celery('worker', broker='redis://localhost:6379/1')

@app.task
def add(pri):
    print "add called with pri: {}".format(pri)
    time.sleep(5)
    return pri

@app.task
def sub(pri):
    print "sub called with pri: {}".format(pri)
    time.sleep(5)
    return pri