# celery 任务队列, celery 定制路由, 改进版

add.py

from celery import Celery
app_add = Celery('app_add',backend='redis://localhost:6379/0',broker='redis://localhost:6379/0')

from kombu import Exchange, Queue

media_exchange = Exchange('app_reduce', type='direct')

app_add.conf.task_queues = (
    Queue('app_reduce', media_exchange, routing_key='app_reduce'),
)
# 启动任务队列时, 可使用缺省队列celery.
# app_add.conf.task_default_queue = 'app_add'
# app_add.conf.task_default_exchange = 'app_add'
# app_add.conf.task_default_routing_key = 'app_add'

@app_add.task
def add(x, y):
    return x + y

call_tasks.py

from add import add
from reduce import reduce

if __name__ == "__main__":
	for i in range(2000):
		reduce.delay(i, i)
	for i in range(2000):
		add.delay(i, i)

reduce.py

from add import  app_add

@app_add.task(queue='app_reduce')
def reduce(x, y):
    return x -y