篇首语：本文由编程笔记#小编为大家整理，主要介绍了celery相关的知识，希望对你有一定的参考价值。

Celery

Celery是一个异步任务框架,是一个独立运行的服务.(内置socket)

相当于一个生产者消费者模型的任务队列.

拥有高可用,异步,简易,等特点.

celery是一个独立的socket

官网

Celery 官网：http://www.celeryproject.org/

Celery 官方文档英文版：http://docs.celeryproject.org/en/latest/index.html

Celery 官方文档中文版：http://docs.jinkan.org/docs/celery/

Celery架构

Celery的架构由三部分组成，消息中间件（message broker）、任务执行单元（worker）和 任务执行结果存储(backend)（task result store）组成。

RabbitMQ 异步消息队列,可编程

Redis 数据库

消息中间件

Celery本身不提供消息服务,但是可以方便的于第三方提供的消息中间件集成.包括RabbitMQ,Redis等.

任务执行单元

Worker是Celery提供的任务执行的单元.worker并发的运行在系统节点中.

任务结果存储

Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, redis等,一般需要支持过期时间.

使用场景

异步执行delay()：解决耗时任务

# 添加任务
# result = add.delay(20, 30)
# print(result.id)

1）创建Celery框架对象app，配置broker和backend，得到的app就是worker
2）给worker对应的app添加可处理的任务函数
3）启动celery服务，运行worker
4）书写添加任务的脚本，执行脚本添加任务到broker，worker会自己异步从broker中拿任务执行，执行结果放在backend中
5) 书写获取任务结果的脚本，明确任务id与执行的app，获取任务结果

延迟执行：解决延迟任务

# 添加延迟任务
from datetime import datetime, timedelta
result = add.apply_async(args=(10, 20), eta=datetime.utcnow() + timedelta(secOnds=10))
print(result)

1）创建Celery框架对象app，配置broker和backend，得到的app就是worker
2）给worker对应的app添加可处理的任务函数，用include配置给worker的app
3）启动celery服务，运行worker
4）书写添加任务的脚本，执行脚本添加任务到broker，worker会自己异步从broker中拿任务执行，执行结果放在backend中
5) 书写获取任务结果的脚本，明确任务id与执行的app，获取任务结果

定时执行：解决周期(周期)任务

# 时区
app.conf.timezOne= &＃39;Asia/Shanghai&＃39;
# 是否使用
UTCapp.conf.enable_utc = False
# 定时任务配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
&＃39;add-task&＃39;: {
&＃39;task&＃39;: &＃39;celery_task.tasks.add&＃39;,
&＃39;schedule&＃39;: timedelta(secOnds=3),
# &＃39;schedule&＃39;: crontab(hour=8, day_of_week=1), # 每周一早八点
&＃39;args&＃39;: (20, 10),
},
&＃39;low-task&＃39;: {
&＃39;task&＃39;: &＃39;celery_task.tasks.low&＃39;,
&＃39;schedule&＃39;: timedelta(secOnds=6),
&＃39;args&＃39;: (20, 10), },
}

1）创建Celery框架对象app，配置broker和backend，得到的app就是worker
2）给worker对应的app添加可处理的任务函数，用include配置给worker的app
3）完成提供的任务的定时配置app.conf.beat_schedule
4）启动celery服务，运行worker，执行任务
5）启动beat服务，运行beat，添加任务

Celery安装与配置

安装

pip install celery

消息中间件

RaddbiMQ/Redis

实例化对象

app = Celert(&＃39;任务名&＃39;,broker="xxx",backend="xxx")

Celery执行异步任务

包架构封装

project
├── celery_task # celery包
│ ├── __init__.py # 包文件
│ ├── celery.py # celery连接和配置相关文件，且名字必须交celery.py
│ └── tasks.py # 所有任务函数
├── add_task.py # 添加任务
└── get_result.py # 获取结果

异步执行

celery.py

# 1）创建app + 任务
# 2）启动celery(app)服务：
# 非windows
# 命令：celery worker -A celery_task -l info
# windows：
# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet
# 3）添加任务：手动添加，要自定义添加任务的脚本，右键执行脚本
# 4）获取结果：手动获取，要自定义获取任务的脚本，右键执行脚本
from celery import Celery
#broker
broker = &＃39;redis://127.0.0.1:6379/1&＃39;
#backend
backend = &＃39;redis://127.0.0.1:6379/2&＃39;
#worker
app = Celery(broker=broker, backend=backend, include=[&＃39;celery_task.tasks&＃39;]) #include是任务tasks

task.py

from .celery import app
import time
@app.task
def add(n, m):
print(n)
print(m)
time.sleep(10)
print(&＃39;n+m的结果：%s&＃39; % (n + m))
return n + m
@app.task
def low(n, m):
print(n)
print(m)
print(&＃39;n-m的结果：%s&＃39; % (n - m))
return n - m

add_task.py

from celery_task import tasks
# 添加立即执行任务
t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1.id)
# 添加延迟任务
from datetime import datetime, timedelta
eta=datetime.utcnow() + timedelta(secOnds=10)
tasks.low.apply_async(args=(200, 50), eta=eta)

get_result.py

from celery_task.celery import app
from celery.result import AsyncResult
id = &＃39;21325a40-9d32-44b5-a701-9a31cc3c74b5&＃39;
if __name__ == &＃39;__main__&＃39;:
async = AsyncResult(id=id, app=app)
if async.successful():
result = async.get()
print(result)
elif async.failed():
print(&＃39;任务失败&＃39;)
elif async.status == &＃39;PENDING&＃39;:
print(&＃39;任务等待中被执行&＃39;)
elif async.status == &＃39;RETRY&＃39;:
print(&＃39;任务异常后正在重试&＃39;)
elif async.status == &＃39;STARTED&＃39;:
print(&＃39;任务已经开始被执行&＃39;)

定时任务

celery.py

# 1）创建app + 任务
# 2）启动celery(app)服务：
# 非windows
# 命令：celery worker -A celery_task -l info
# windows：
# pip3 install eventlet
# celery worker -A celery_task -l info -P eventlet
# 3）添加任务：自动添加任务，所以要启动一个添加任务的服务
# 命令：celery beat -A celery_task -l info
# 4）获取结果
from celery import Celery
broker = &＃39;redis://127.0.0.1:6379/1&＃39;
backend = &＃39;redis://127.0.0.1:6379/2&＃39;
app = Celery(broker=broker, backend=backend, include=[&＃39;celery_task.tasks&＃39;])
# 时区
app.conf.timezOne= &＃39;Asia/Shanghai&＃39;
# 是否使用UTC
app.conf.enable_utc = False
# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
&＃39;low-task&＃39;: {
&＃39;task&＃39;: &＃39;celery_task.tasks.low&＃39;,
&＃39;schedule&＃39;: timedelta(secOnds=3),
# &＃39;schedule&＃39;: crontab(hour=8, day_of_week=1), # 每周一早八点
&＃39;args&＃39;: (300, 150),
}
}

tasks.py

from .celery import app
import time
@app.task
def add(n, m):
print(n)
print(m)
time.sleep(10)
print(&＃39;n+m的结果：%s&＃39; % (n + m))
return n + m
@app.task
def low(n, m):
print(n)
print(m)
print(&＃39;n-m的结果：%s&＃39; % (n - m))
return n - m

get_result.py

from celery_task.celery import app
from celery.result import AsyncResult
id = &＃39;21325a40-9d32-44b5-a701-9a31cc3c74b5&＃39;
if __name__ == &＃39;__main__&＃39;:
async = AsyncResult(id=id, app=app)
if async.successful():
result = async.get()
print(result)
elif async.failed():
print(&＃39;任务失败&＃39;)
elif async.status == &＃39;PENDING&＃39;:
print(&＃39;任务等待中被执行&＃39;)
elif async.status == &＃39;RETRY&＃39;:
print(&＃39;任务异常后正在重试&＃39;)
elif async.status == &＃39;STARTED&＃39;:
print(&＃39;任务已经开始被执行&＃39;)