目录
一、celery介绍架构和安装
Celery 官网：http://www.celeryproject.org/
Celery 官方文档英文版：http://docs.celeryproject.org/en/latest/index.html
Celery 官方文档中文版：http://docs.jinkan.org/docs/celery/
1.1 介绍
# celery :分布式的异步任务框架，主要用来做：
- 异步任务
- 延迟任务
- 定时任务--->如果只想做定时任务，可以不使用celery，有别的选择
# celery 框架，原理
1）可以不依赖任何服务器，通过自身命令，启动服务(内部支持socket)
2）celery服务为为其他项目服务提供异步解决任务需求的
注：会有两个服务同时运行，一个是项目服务，一个是celery服务，项目服务将需要异步处理的任务交给celery服务，celery就会在需要时异步完成项目的需求
人是一个独立运行的服务 | 医院也是一个独立运行的服务
正常情况下，人可以完成所有健康情况的动作，不需要医院的参与；但当人生病时，就会被医院接收，解决人生病问题
人生病的处理方案交给医院来解决，所有人不生病时，医院独立运行，人生病时，医院就来解决人生病的需求
# 扩展：分布式的任务部署：celery跟django服务是独立的，代码是都写在一起的。celery是分布式的异步任务框架，那么celery是可以放在多台机器上，在另一台机器上，把django所有的代码都写上(不要只拿celery项目的包)，只启动celery服务，不需要启动django服务。这样原本的机器上把所有的任务都提交给消息队列，两个celery服务分别去消息队列中取任务来执行。起了多个worker，都是从消息队列中取任务，不会产生错乱也不会执行多次。提交了异步任务后，可以运行在分布式的节点上。
1.2 celery架构
- 消息中间件（broker）：消息队列：可以使用redis，rabbitmq，咱们使用redis
- 任务执行单元（worker）：真正的执行 提交的任务
- 任务执行结果存储（banckend）：可以使用mysql，redis，咱们使用redis
## 1.3 安装celery
```python
-pip install Celery
-释放出可执行文件：celery，由于 python解释器的script文件夹再环境变量，任意路径下执行celery都能找到
# celery不支持win，所以想再win上运行，需要额外安装eventlet
windows系统需要eventlet支持：pip install eventlet
Linux与MacOS直接执行：
3.x，4.x版本：celery worker -A demo -l info
5.x版本： celery -A demo worker -l info -P eventlet
二、celery执行异步任务
# 基本使用
1 再虚拟环境中装celery和eventlet
2 写个py文件，实例化得到app对象，注册任务
from celery import Celery
import time
broker = 'redis://127.0.0.1:6379/1' # 消息中间件 redis
backend = 'redis://127.0.0.1:6379/2' # 结果存储 redis
app = Celery(__name__,
broker=broker,
backend=backend)
@app.task # 变成celery的任务了
def add(a, b):
print('运算结果是',a + b)
time.sleep(1)
return a + b
3 启动worker（worker监听消息队列，等待别人提交任务，如果没有就卡再这）
celery -A 启动文件名 worker -l info -P eventlet
celery -A demo worker -l info -P eventlet
4 别人 提交任务，提交完成会返回一个id号，后期使用id号查询，至于这个任务有没有被执行，取决于worker有没有启动
from demo import add
res=add.delay(77,66)
5 提交任务的人，再查看结果
from demo import app
# celery的包下
from celery.result import AsyncResult
id = 'e75fc254-0472-43a3-ba9b-f6edfdd64b06'
if __name__ == '__main__':
asy = AsyncResult(id=id, app=app)
if asy.successful(): # 正常执行完成
result = asy.get() # 任务返回的结果
print(result)
elif asy.failed():
print('任务失败')
elif asy.status == 'PENDING':
print('任务等待中被执行')
elif asy.status == 'RETRY':
print('任务异常后正在重试')
elif asy.status == 'STARTED':
print('任务已经开始被执行')
代码：
scripts/t_celery_demo/demo.py
from celery import Celery
import time
broker = 'redis://127.0.0.1:6379/1' # 消息中间件 redis
backend = 'redis://127.0.0.1:6379/2' # 结果存储 redis
'''
app = Celery()
参数：
main：要和文件的名字一致，是task名字的一部分,
broker：存储接收任务的数据库或mq
backend：存储任务处理结果的数据库或mq
include：适用于寻找目录中所有的task
'''
app = Celery(__name__, broker=broker, backend=backend)
@app.task # 加上这个装饰器，任务就变成celery的任务了
def add(a, b):
print('运算结果是', a + b)
time.sleep(1)
return a + b
# 改动了celery绑定的启动文件后，后续的改动worker不能监听，所以要退出后重新启动worker
@app.task
def send_sms(mobile, code):
time.sleep(1)
print('%s手机号，发送短信成功，验证码为：%s' % (mobile, code))
return True
结果
(luffy) E:\python project\luffy_api\scripts\t_celery_demo>celery -A demo worker -l info -P eventlet
-------------- celery@LAPTOP-EHK3KOOL v5.3.1 (emerald-rush)
--- ***** -----
-- ******* ---- Windows-10-10.0.19045-SP0 2023-06-29 15:16:47
- *** --- * ---
- ** ---------- [config]
- ** ---------- .> app: demo:0x22f03fd68e0
- ** ---------- .> transport: redis://127.0.0.1:6379/1
- ** ---------- .> results: redis://127.0.0.1:6379/2
- *** --- * --- .> concurrency: 4 (eventlet)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
-------------- [queues]
.> celery exchange=celery(direct) key=celery
[tasks]
. demo.add
. demo.send_sms
[2023-06-29 15:40:12,840: INFO/MainProcess] pidbox: Connected to redis://127.0.0.1:6379/1.
[2023-06-29 15:40:56,739: INFO/MainProcess] Task demo.send_sms[726ae408-9e94-4ebc-af53-424037014e25]
received
[2023-06-29 15:40:57,760: WARNING/MainProcess] 17903672785手机号，发送短信成功，验证码为：8888
[2023-06-29 15:40:57,784: INFO/MainProcess] Task demo.send_sms[726ae408-9e94-4ebc-af53-424037014e25]
succeeded in 1.0470000002533197s: True
ved
[2023-06-29 15:42:04,816: WARNING/MainProcess] 运算结果是
[2023-06-29 15:42:04,817: WARNING/MainProcess]
[2023-06-29 15:42:04,817: WARNING/MainProcess] 143
[2023-06-29 15:42:05,834: INFO/MainProcess] Task demo.add[4ca12018-839
2-4608-a00e-447dcabcd65a] succeeded in 1.0159999998286366s: 143
scripts/t_celery_demo/add_task.py
from demo import add # 直接引用函数
# # 同步运行，同步调用
# res = add(4, 5)
# print(res)
## 使用celery异步调用
# 步骤1：异步调用--->执行下面，实际上是：把任务提交到消息队列中，返回一个id号，后期通过id号查询任务执行结果，并没有执行，得等待worker执行
res = add.delay(77, 66)
print(res) # e75fc254-0472-43a3-ba9b-f6edfdd64b06
# 步骤2：启动worker
# pip install eventlet
# 到执行文件的路径下：E:\python project\luffy_api\scripts\t_celery_demo>
# celery -A demo worker -l info -P eventlet # 启动了celery服务，以后不用启动了，只需要提交任务，worker会自动执行
# 步骤3：去结果中查看是否执行完成--get_result.py
scripts/t_celery_demo/add_task2.py
from demo import send_sms
res = send_sms.delay(17903672785, 8888)
print(res) # 726ae408-9e94-4ebc-af53-424037014e25
scripts/t_celery_demo/get_result.py
from demo import app
# celery的包下
from celery.result import AsyncResult
id = '726ae408-9e94-4ebc-af53-424037014e25'
if __name__ == '__main__':
asy = AsyncResult(id=id, app=app)
if asy.successful(): # 正常执行完成
result = asy.get() # 任务返回的结果
print(result)
elif asy.failed():
print('任务失败')
elif asy.status == 'PENDING':
print('任务等待中被执行')
elif asy.status == 'RETRY':
print('任务异常后正在重试')
elif asy.status == 'STARTED':
print('任务已经开始被执行')
三、包架构封装
scripts
├── celery_task # celery包
│ ├── __init__.py # 包文件
│ ├── celery.py # celery连接和配置相关文件，且名字必须交celery.py
│ ├── course_task.py # 课程任务函数
│ ├── home_task.py # 首页任务函数
│ ├── user_task.py # 用户任务函数
├── add_task.py # 添加任务
└── get_result.py # 获取结果
使用步骤：
1 新建包：celery_task
2 再包下新建 celery.py 必须叫它，里面实例化得到app对象
from celery import Celery
broker = 'redis://127.0.0.1:6379/1' # 消息中间件 redis
backend = 'redis://127.0.0.1:6379/2' # 结果存储 redis
app = Celery(__name__, broker=broker, backend=backend, include=['celery_task.course_task','celery_task.home_task','celery_task.user_task'])
3 新建任务py文件：user_task.py course_task.py home_task.py
-以后跟谁相关的任务，就写在谁里面
from .celery import app
import time
@app.task
def send_sms(mobile, code):
time.sleep(2)
print('%s手机号，发送短信成功，验证码是：%s' % (mobile, code))
return True
4 启动worker，以包启动，来到包所在路径下
celery -A 包名 worker -l info -P eventlet
celery -A celery_task worker -l info -P eventlet
5 其它程序，导入任务，提交任务即可
from celery_task.user_task import send_sms
res = send_sms.delay(1999999333, 8888)
print(res) # f33ba3c5-9b78-467a-94d6-17b9074e8533
6 其它程序，查询结果
from celery_task.celery import app
# celery的包下
from celery.result import AsyncResult
id = '51a669a3-c96c-4f8c-a5fc-e1b8e2189ed0'
if __name__ == '__main__':
a = AsyncResult(id=id, app=app)
if a.successful(): # 正常执行完成
result = a.get() # 任务返回的结果
print(result)
elif a.failed():
print('任务失败')
elif a.status == 'PENDING':
print('任务等待中被执行')
elif a.status == 'RETRY':
print('任务异常后正在重试')
elif a.status == 'STARTED':
print('任务已经开始被执行')
四、celery执行延迟任务和定时任务
# celery 可以做
-异步任务
-延迟任务--->延迟多长时间干任务
-定时任务：每天12点钟，每隔几秒...
-如果只做定时任务，不需要使用celery这么重，apscheduler（自己去研究）
# 异步任务
-导入异步任务的函数
-函数.delay(参数)
# 延迟任务
-导入异步任务的函数
-函数.apply_async(kwargs={'mobile':'1896334234','code':8888},eta=时间对象)
# 定时任务：在app所在的文件下配置
- 1 配置
app.conf.beat_schedule = {
'send_sms': {
'task': 'celery_task.user_task.send_sms',
'schedule': timedelta(seconds=5),
'args': ('1822344343', 8888),
},
'add_course': {
'task': 'celery_task.course_task.add_course',
# 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点
'schedule': crontab(hour=11, minute=38), # 每天11点35，执行
'args': (),
}
}
-2 启动beat，启动worker
celery -A celery_task worker -l info -P eventlet
celery -A celery_task beat -l info
-3 到了时间，beat进程负责提交任务到消息队列--->worker执行
异步任务
from celery_task.user_task import send_sms
##### 1.异步任务
res = send_sms.delay(18955667325, 8888)
print(res) # 81ece605-b715-4f77-9218-f883478e8333
延迟任务
### 时间对象
from datetime import datetime, timedelta
print(datetime.now()) # 当前时间，做了时间国际化，就可以使用这个了
# 2023-06-29 16:34:15.430620
print(datetime.utcnow()) # 咱们写延迟任务要用utc时间，没有做时间国际化
# 2023-06-29 08:34:15.430620
print(type(datetime.utcnow()))
# <class 'datetime.datetime'> 时间对象
# days=0, seconds=0, microseconds=0, milliseconds=0, minutes=0, hours=0, weeks=0
print(timedelta(hours=2, minutes=26, seconds=7)) # 2:26:07
print(type(timedelta(hours=2))) # <class 'datetime.timedelta'> 时间对象
# 这两个对象可以相加
eta = datetime.utcnow() + timedelta(minutes=2)
print(eta) # 2023-06-29 08:40:53.191965
### 2.延时任务
# res = send_sms.apply_async(args=['1896334234', 8888])
# print(res) # a310f745-5903-46a2-b328-b240130b29e6
# 延时任务
from datetime import datetime, timedelta
eta = datetime.utcnow() + timedelta(minutes=2)
res = send_sms.apply_async(kwargs={'mobile': '1896334234', 'code': 8888}, eta=eta)
print(res) # 417d3cfe-5a8c-44ed-b802-7c658c4e655e
结果是：
[2023-06-29 17:22:10,489: INFO/MainProcess] Task celery_task.user_task.send_sms[d1bc1685-c035-455a-840e-4aac0113cd24] received # 收到了任务，延时结束后才会执行
[2023-06-29 17:22:14,803: WARNING/MainProcess]1896334234手机号，发送短信成功，验证码是：8888 # 真正执行任务
[2023-06-29 17:22:14,846: INFO/MainProcess] Task celery_task.user_task.send_sms[4beb5510-9d5d-4af8-a7f7-e850e45270f8] succeeded in 2.0619999999180436s: True
定时任务
# 3 定时任务 1 每隔5s干一次 2 每天的几点几分几秒干一次
from celery import Celery
broker = 'redis://127.0.0.1:6379/1' # 消息中间件 redis
backend = 'redis://127.0.0.1:6379/2' # 结果存储 redis
# include是列表包下的任务文件
app = Celery(__name__, broker=broker, backend=backend,
include=['celery_task.course_task', 'celery_task.home_task', 'celery_task.user_task'])
# 定时任务写在celery中
# app.conf是app的所有配置
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
'send_sms': {
'task': 'celery_task.user_task.send_sms', # 任务关联的函数
'schedule': timedelta(seconds=5), # 每隔5秒发一次
'args': ('1822344343', 8888), # 函数的参数
},
'add_course': { # 增加课程
'task': 'celery_task.course_task.add_course',
# 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点
'schedule': crontab(hour=19, minute=26), # 每天11点35，执行
'args': (),
}
}
-2 启动beat，启动worker
celery -A celery_task worker -l info -P eventlet
celery -A celery_task beat -l info
-3 到了时间，beat进程负责提交任务到消息队列--->worker执行
五、django中使用celery
# 使用步骤
1 把之前写好的包，copy到项目根路径下
2 在xx_task.py 中写任务
from .celery import app
@app.task
def add_banner():
from home.models import Banner
Banner.objects.create(title='测试', image='/1.png', link='/banner', info='xxx',orders=99)
return 'banner增加成功'
3 在celery.py 中加载django配置
# 一、加载django配置环境
import os
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffy_api.settings.dev")
4 视图函数中，导入任务，提交即可
from celery_task.home_task import add_banner
class CeleryView(APIView):
def get(self, request):
res = add_banner.delay()
return APIResponse(msg='新增banner的任务已经提交了')
5 启动worker，等待运行即可
5.1 解释
#1 celery中要使用djagno的东西，才要加这句话
import os
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffy_api.settings.dev")
user/views.py
##### 测试celery的使用---> 前端发送一个get请求，就异步向banner表中插入一条记录
from celery_task.home_task import add_banner
class CeleryView(APIView):
def get(self, request):
# 同步使用，我们不使用这个
# add_banner()
# 异步
res = add_banner.delay()
print(res)
return APIResponse(msg='新增banner的任务已经提交了')
启动celery
到项目路径下
E:\python project\luffy_api>celery -A celery_task worker -l info -P eventlet