转：CELERY CELERY_QUEUES和CELERY_ROUTS的用法

标签：celery task2 task1 QUEUES queue CELERY ROUTS app

转自：https://www.jianshu.com/p/4d0bbdbc6ade?utm_campaign=maleskine&utm_content=note&utm_medium=seo_notes&utm_source=recommendation

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

 

标签：celery,task2,task1,QUEUES,queue,CELERY,ROUTS,app
From： https://www.cnblogs.com/xuxiaobo/p/18666603