celery

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



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