celery:

1.

参考文档：

https://www.jianshu.com/p/4d0bbdbc6ade

http://docs.jinkan.org/docs/celery/

简介：

celery对象+配置

在任务上加上decorator

不同的任务放入不同的队列

定义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

2.

图示

C_R C_Q

celery_producer-->exchange-->queue-->celery__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

-------------------------------------------

to_create.apply_async(args=[data_str],retry=True,queue='to_create',immutable=True)

参考文档：

http://docs.jinkan.org/docs/celery/reference/celery.app.task.html

#args - 传递给任务的位置参数（列表或元组）。

args – The positional arguments to pass on to the task (a list or tuple).

#重试 - 如果启用，则在连接丢失或失败时将重试发送任务消息。 默认值取自CELERY_TASK_PUBLISH_RETRY设置。 请注意，您需要手动处理生产者/连接才能使其正常工作。

retry – If enabled sending of the task message will be retried in the event of connection loss or failure. Default is taken from the CELERY_TASK_PUBLISH_RETRY setting. Note you need to handle the producer/connection manually for this to work.

#queue - 将任务路由到的队列。 这必须是CELERY_QUEUES中的键，或者必须启用CELERY_CREATE_MISSING_QUEUES。 有关更多信息，请参阅路由任务

queue – The queue to route the task to. This must be a key present in CELERY_QUEUES, or CELERY_CREATE_MISSING_QUEUES must be enabled. See Routing Tasks for more information.

# Immutable不希望带上上一个任务的结果

immutable=True

https://www.jianshu.com/p/d401caab0a7a

-------------------------------------------