# celery 任务队列, 不同celery实例定制不同的默认队列及路由设定.

add.py

# 说明文档
# https://docs.celeryproject.org/en/latest/userguide/routing.html#broadcast
from celery import Celery
app_add = Celery('app_add',backend='redis://localhost:6379/0',broker='redis://localhost:6379/0')

from kombu import Exchange, Queue

# default_exchange = Exchange('default', type='direct')
# media_exchange = Exchange('media', type='direct')
#
# app_add.conf.task_queues = (
#     Queue('default', default_exchange, routing_key='default'),
#     Queue('videos', media_exchange, routing_key='media.video'),
#     Queue('images', media_exchange, routing_key='media.image')
# )
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_task.py

from add import add
from reduce import reduce
"""
app.conf.CELERY_ROUTES = {
    "celery_app.hello": {
        "queue": "test"
    }
}
from kombu import Exchange, Queue
"""
if __name__ == "__main__":
	for i in range(2000):
		reduce.delay(i, i)
	for i in range(2000):
		add.delay(i, i)

reduce.py

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

app_reduce.conf.task_default_queue = 'app_reduce'
app_reduce.conf.task_default_exchange = 'app_reduce'
app_reduce.conf.task_default_routing_key = 'app_reduce'

@app_reduce.task
def reduce(x, y):
    return x -y

优化方向, 启动任务指定任务队列