1. 介绍
Celery非常容易设置，通常都是使用默认的queue来存放任务，写法如下:
@app.task
def task1(x, y):
for _ in range(10):
time.sleep(1)
print('x + y =', x + y)
return x + y
@app.task
def task2():
for _ in range(100):
print('task2: ', datetime.now())
time.sleep(1)
这两个任务都会在同一个queue里面执行，这样写很简单，因为只需要一个decorator就能实现一个异步任务。但如果考虑到每个任务的执行时间耗费资源或者重要程度不同，把两个任务放到同一个queue中，可能造成执行时间长但重要程度却低的任务先执行，极大的影响程序的功能。在使用同一个queue时，就算增加worker也无法解决该问题。
为了解决该问题，需要把task1放到queue1中，把task2放到queue2中去执行。同时指定worker1去处理queue1的任务，worker2去处理queue2的任务。使用这种方式时，各个任务就能获得足够的worker去处理，同时一些优先级worker也能很好的处理重要的任务而不需要等待了。
2. 定义queue和routes
首先手动定义queue
CELERY_QUEUES = (
Queue('default', exchange=Exchange('default'), routing_key='default'),
Queue('app_task1', exchange=Exchange('app_task1'), routing_key='app_task1'),
Queue('app_task2', exchange=Exchange('app_task2'), routing_key='app_task2'),
)
然后定义routes用来决定不同的任务去哪一个queue
CELERY_ROUTES = {
'celery_app.task.task1': {'queue': 'app_task1', 'routing_key': 'app_task1'},
'celery_app.task.task2': {'queue': 'app_task2', 'routing_key': 'app_task2'}
}
在启动worker时指定该worker执行哪一个queue中的任务
celery -A celery_app worker -l info -Q app_task1 -P eventlet
celery -A celery_app worker -l info -Q app_task2 -P eventlet
举例：
如果某项目中涉及到大量文件转换问题，有大量小于1MB的文件转换，同时也有少量的20MB的文件转换，小文件转换的优先级是最高的，同时不用占用太多时间，但大文件的转换很耗时。如果将转换任务放到一个队列里面，那么很有可能因为出现转换大文件，导致耗时太严重造成小文件转换延时的问题。
所以可以按照文件大小设置3个优先队列，并且每个队列设置不用的worker。
自己的理解，若有问题日后修正:
celery的生产者会根据CELERY_ROUTES的值，将不同的任务放到不同的Exchange中，exchange根据CELERY_QUEUES的值将任务分配到不同的queue中，在worker指定取任务的queue后，那么就只从该queue中取出任务然后执行。
3. 代码
celery_app
__init__.py
celeryconfig.py
main.py
task.py
3.1 init.py
from celery import Celery
app = Celery('celery_app') # include=['celery_app.task']
app.config_from_object('celery_app.celeryconfig')
3.2 celeryconfig.py
配置文件
from kombu import Queue, Exchange
BROKER_URL = 'redis://127.0.0.1:6379/7'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/8'
CELERY_IMPORTS = (
'celery_app.task'
)
CELERY_QUEUES = (
Queue('default', exchange=Exchange('default'), routing_key='default'),
Queue('app_task1', exchange=Exchange('app_task1'), routing_key='app_task1'),
Queue('app_task2', exchange=Exchange('app_task2'), routing_key='app_task2'),
)
CELERY_ROUTES = {
'celery_app.task.task1': {'queue': 'app_task1', 'routing_key': 'app_task1'},
'celery_app.task.task2': {'queue': 'app_task2', 'routing_key': 'app_task2'}
}
3.3 task.py
具体任务
import time
from datetime import datetime
from celery_app import app
@app.task
def task1(x, y):
for _ in range(10):
time.sleep(1)
print('x + y =', x + y)
return x + y
@app.task
def task2():
for _ in range(100):
print('task2: ', datetime.now())
time.sleep(1)
3.4 main.py
执行任务
from celery_app.task import task1, task2
r = task1.apply_async(args=(1, 2))
r2 = task2.delay()
print(r.status)
print(r2.status)
3.5 执行
先启动虚拟环境，执行
celery -A celery_app worker -l info -Q app_task1 -P eventlet
# 或者
celery -A celery_app worker -l info -Q app_task2 -P eventlet
然后执行main.py文件
就可以看到两个worker分别执行不同的任务，并且只会执行被分配的任务了。
4. 注意事项
4.1 CELERY_IMPORTS问题
CELERY_IMPORTS = (
'celery_app.task'
)
这个属性中配置的是需要执行的任务的模块，如果没有配置，那么在启动worker之后，便会报错，因为CELERY_ROUTES中的任务将会无法找到。
或者不想配置这个，也可以在创建Celery对象时传入参数配置，
app = Celery('celery_app',include=['celery_app.task'])
4.2 CELERY_ROUTES
CELERY_ROUTES的作用是，给任务分配queue和routing_key，然后根据给worker分配的queue值执行相应的任务。
如果在celeryconfig.py中没有配置该项，那么也可以这么写，
启动worker:
celery -A celery_app worker -l info -Q app_task1 -P eventlet
然后在生产任务时，主动传入queue和routing_key的值
r = task1.apply_async(args=(1, 2), queue='app_task1', routing_key='app_task1')
4.3 Exchange
如果在使用redis做BROKEN时，在创建Queue对象时，其实可以不用传入Exchange的值，即
CELERY_QUEUES = (
Queue('default', routing_key='default'),
Queue('app_task1', routing_key='app_task1'),
Queue('app_task2', routing_key='app_task2'),
)
但如果使用了的是RabbitMQ，那么这个值就一定需要。
所以以防以后更改了BROKEN程序失效，那么在配置Queue时，默认将这个参数传入，然后值跟Queue的名字一样即可。
4.4 queue和routing_key
这两个值的名字不需要保持一致，那么为了方便使用和检查，最好还是保持一致。
4.5 定时任务
在上面添加代码
CELERYBEAT_SCHEDULE = {
'celery_app.task.task1': {
'task': 'celery_app.task.task1',
'schedule': timedelta(seconds=20),
'args': (1, 10)
},
'celery_app.task.task2': {
'task': 'celery_app.task.task2',
'schedule': crontab(minute='*/2'),
'args': ()
}
}
属性名称不要写错了，是CELERYBEAT_SCHEDULE，不要写成了BEAT_SCHEDULE或者CELERY_BEAT_SCHEDULE了。
启动定时器:
CELERY -A celery_app beat
在启动worker时，也可以指定queue，那么该worker就只执行该queue中的任务。
CELERY -A celery_app worker -l info -Q app_task1 -P eventlet
5. 参考:
https://denibertovic.com/posts/celery-best-practices/
https://blog.csdn.net/siddontang/article/details/34447003