一、Celery异步分布式
Celery 是一个python开发的异步分布式任务调度模块，是一个消息传输的中间件，可以理解为一个邮箱，每当应用程序调用celery的异步任务时，会向broker传递消息，然后celery的worker从中取消息
Celery 用于存储消息以及celery执行的一些消息和结果
对于brokers，官方推荐是rabbitmq和redis
对于backend，也就是指数据库，为了简单一般使用redis
使用redis连接url格式：
redis://:[email protected]:port/db_number
1）定义连接脚本tasks.py
|
1
2
3
4
5
6
7
8
9
|
#!/usr/bin/env pythonfrom celery import Celery
broker = "redis://192.168.2.230:6379/1"
backend = "redis://192.168.2.230:6379/2"
app = Celery("tasks", broker=broker, backend=backend)
@app.task
def add(x,y):
return x+y
|
2）安装启动celery
pip install celery
pip install redis
启动方式：celery -A huang tasks -l info #-l 等同于 --loglevel
3）执行测试 huang.py
|
1
2
3
4
5
6
7
8
9
10
|
#!/usr/bin/env pythonfrom tasks import add
re = add.delay(10,20)
print(re.result) #任务返回值
print(re.ready) #如果任务被执行返回True,其他情况返回False
print(re.get(timeout=1)) #带参数的等待，最后返回结果
print(re.status) #任务当前状态
|
运行结果：
30
<bound method AsyncResult.ready of <AsyncResult: d2e0a2d8-cdd9-4fe3-a8bb-81fe3c53ba9a>>
30
SUCCESS
4）根据成功返回的key或celery界面输出的信息，查看redis存储
说明：停止celery服务，执行完huang.py之后，再启动celery服务也是有保存数据的
二、celery多进程
1）配置文件 celeryconfig.py
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
#!/usr/bin/env python#-*- coding:utf-8 -*-from kombu import Exchange,Queue
BROKER_URL = "redis://192.168.2.230:6379/3"
CELERY_RESULT_BACKEND = "redis://192.168.2.230:6379/4"
CELERY_QUEUES = (
Queue("default",Exchange("default"),routing_key="default"),
Queue("for_task_A",Exchange("for_task_A"),routing_key="for_task_A"),
Queue("for_task_B",Exchange("for_task_B"),routing_key="for_task_B")
)CELERY_ROUTES = {
'tasks.taskA':{"queue":"for_task_A","routing_key":"for_task_A"},
'tasks.taskB':{"queue":"for_task_B","routing_key":"for_task_B"}
} |
2）tasks.py
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
#!/usr/bin/env python#-*- coding:utf-8 -*-from celery import Celery
app = Celery()
app.config_from_object("celeryconfig")
@app.task
def taskA(x,y):
return x+y
@app.task
def taskB(x,y,z):
return x+y+z
|
3）启动celery
celery -A tasks worker --loglevel info
4）执行脚本huang2.py
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
#!/usr/bin/env python#-*- coding:utf-8 -*-from tasks import taskA,taskB
re = taskA.delay(10,20)
print(re.result) #任务返回值
print(re.ready) #如果任务被执行返回True,其他情况返回False
print(re.get(timeout=1)) #带参数的等待，最后返回结果
print(re.status) #任务当前状态
re2 = taskB.delay(10,20,30)
print(re2.result)
print(re2.ready)
print(re2.get(timeout=1))
print(re2.status)
|
5）运行结果
None
<bound method AsyncResult.ready of <AsyncResult: e34a8490-05a7-473e-a082-f4956cabfc99>>
30
SUCCESS
None
<bound method AsyncResult.ready of <AsyncResult: 3c5cd839-dbe2-4e63-ba4e-86e8c79d943f>>
60
SUCCESS