目录
Celery介绍、安装、基本使用
一、Celery服务
什么是Celery:
Celery是一个简单、灵活且可靠的，处理消息的分布式系统
- Celery可以用来做什么：
- 异步任务
- 定时任务
- 延迟任务
Celery的运行原理:
- 可以不依赖任何服务，通过自身命令，启动服务
- celery服务为其他项目服务提供异步解决任务需求
# 注：会有两个服务同时运行 - 项目服务 - celery服务 项目服务将需要异步处理的任务交给celery服务，celery就会在需要时异步完成项目的需求 ''' 人是一个独立运行的服务 | 医院也是一个独立运行的服务 正常情况下，人可以完成所有健康情况的动作，不需要医院的参与；但当人生病时，就会被医院接收，解决人生病问题 人生病的处理方案交给医院来解决，所有人不生病时，医院独立运行，人生病时，医院就来解决人生病的需求 ''' 1、celery架构
-
消息中间件：broker
- 提交的任务【函数】都放在这里， celery本身不能提供消息中间件
- 需要借助于第三方： redis或rabbitmq
-
任务执行单元：worker
- 真正执行任务的的地方，一个个进程中执行函数
-
结果储存：backend
- 函数return的结果都存储在这里， celery本身不提供结果存储
- 需要借助于第三方: redis或rabbitmq
使用场景：
- 异步执行：解决耗时任务
- 延迟执行：解决延迟任务
- 定时执行：解决周期任务
2、celery快速使用
Celery不支持在windows上直接运行，通过eventlet支持在win上运行
安装：
pip install celery pip install eventlet # windows需要安装 快速使用：
- 1、第一步：创建一个py文件（main.py），用于实例化celery对象，编写需要执行的函数 # 1、导入模块 from celery import Celery # 2、指定briker，用于存放提交的异步任务 broker = 'redis://127.0.0.1:6379/1' # 3、指定backend，用于存放函数执行结束的结果 backend = 'redis://127.0.0.1:6379/2' # 实例化celery对象 app = Celery('test', broker=broker, backend=backend) # 编写一个函数，装饰上celery对象 @app.task def add(a, b): import time time.sleep(3) print('add函数执行完成') return a + b - 2、第二步：再次创建一个py文件（run.py）,用于将函数提交给celery # 1、导入刚才编写的函数 from main import add # 2、将任务提交给broker,函数需要的参数需要传入 res = add.delay(1, 2) # 3、提交后可以获得该任务的ID，可通过ID可以查询任务执行结果 print(res) # 0213d2c2-453e-41a8-a171-e31f1f2f4883 - 3、第三步：使用命令开启worker （也可以提前开启，任务提交后就会直接执行） # 启动worker命令，win需要安装eventlet # 启动需要进入main.py文件的目录下 win: -4.x之前版本 celery worker -A main -l info -P eventlet -4.x之后 celery -A main worker -l info -P eventlet mac： celery -A main worker -l info - 4、第四步：worker会将执行的结果存在之前指定的broker目录下（指定的redis数据库） - 5、第五步：通过代码查看执行结果（创建新的py文件，专门用于查看执行结果） # 1、导入celery实例的对象 from main import app # 2、导入该模块用于查看结果 from celery.result import AsyncResult # 3、将提交的任务编号拿过来，用于查询结果 id = '0213d2c2-453e-41a8-a171-e31f1f2f4883' # 4、指定该文件为启动文件 if __name__ == '__main__': # 实例化对象，将任务的ID和celery实例化对象当作参数传入 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('任务已经开始被执行')
二、Celer包结构
1、创建clery包结构
什么是包结构：通过将celery服务封装成包的形式，放在项目需要使用的时候导入即可
project ├── celery_task # celery包 │ ├── __init__.py # 包文件 │ ├── celery.py # celery连接和配置相关文件，且名字必须交celery.py │ └── tasks.py # 所有任务函数 ├── add_task.py # 添加任务 └── get_result.py # 获取结果 创建包：
创建一个包，名为：celery_task
- 1、第一步：在包下创建py文件（名字必须为celery.py） # 导入celery模块 from celery import Celery # 导入配置broker和backend from .settings import BACKEND, BROKER # 实例化celery对象 app = Celery('test', broker=BROKER, backend=BACKEND, include=['celery_task.order_task', 'celery_task.user_task']) - 2、第二步：创建settings.py，用于存放配置 BROKER = 'redis://127.0.0.1:6379/1' BACKEND = 'redis://127.0.0.1:6379/2' - 3、第三步，创建py文件（task.py），用于存放需要执行的异步任务 # 导入celery实例对象 from .celery import app # 计算函数 @app.task() def add(a, b): print('计算结果为:', a + b) return True # 模拟发送短信 @app.task() def send_sms(mobile, code): print('已向手机号：%s 发送短信，验证码为：%s' % (mobile, code)) return True - 4、第四步：开启worker 切换到celery所在的目录下，开启worker命令 celery -A celery_task worker -l info -P eventlet - 5、第五步：提桥任务： # add_task.py 文件下 # 提交任务，这里模拟的是异步任务的提交 res = add.delay(a, b) # 提交后可以接收任务的ID res1 = send_sms.delay(mobile, code) - 6、第六步：查看任务执行结果： # get_result.py 文件下 # 导入celery实例 from celery_task.celery import app from celery.result import AsyncResult id = res id1 = res1 # 通过传入任务的ID就可以查询到任务的执行结果 def res_func(id): id = id a = AsyncResult(id=id, app=app) if a.successful(): # 执行完了 result = a.get() if result: return '执行完成' elif a.failed(): return '任务失败,失败的原因可能是未开启worker' elif a.status == 'PENDING': return '任务等待中被执行，当前任务较多或未开启worker' elif a.status == 'RETRY': return '任务异常后正在重试' elif a.status == 'STARTED': return '任务已经开始被执行，请稍后查询' 2、Celery执行异步任务、延迟任务、定时任务
执行异步任务：
# 代码用法： 函数名.delay('函数执行需要的参数') res = func.delay(*args,**kwargs) # res 用于接收提交任务的ID 执行延迟任务：
# 代码用法： # 1、执行延迟任务 from datetime import datetime, timedelta # 设置延迟后的时间，一分钟后执行 eat = datetime.utcnow() + timedelta(minutes=1) # 提交任务 res = send_sms.apply_async(args=['13855411111', '123'], eta=eta) 执行定时任务：
执行定时任务需要启动beat和worker
- beat：定时提交任务的进程---》配置在app.conf.beat_schedule的任务
- worker：执行任务
- 第一步：在celery的py文件中写入 # 导入定时需要的模块 from celery.schedules import crontab # 第一步：在celery的py文件中写入 app.conf.timezone = 'Asia/Shanghai' # 是否使用UTC app.conf.enable_utc = False # celery的配置文件##### # 任务的定时配置 app.conf.beat_schedule = { 'send_sms': { # 配置执行函数的名字 'task': 'celery_task.task.send_sms', # 导入任务的位置 # 'schedule': timedelta(seconds=3), # 时间对象 # 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点 'schedule': crontab(hour=9, minute=43), # 每天9点43 'args': ('18888888', '6666'), # 配置执行函数需要的参数 }, } - 第二步：启动beat # 启动后配配置的任务会自动提交 celery -A celery_task beat -l info - 第三步：启动worker # beat提交的任务被被执行 celery -A celery_task worker -l info -P eventlet 三、Django中使用celery
补充：
如果在公司中，只做定时任务有一个框架更简单一点
使用步骤：
-1 把咱们写的包，复制到项目目录下 -luffy_api -celery_task #celery的包路径 -luffy_api #源代码路径 -2 在使用提交异步任务的位置，导入使用即可 -视图函数中使用，导入任务 -任务.delay() # 提交任务 -3 启动worker，如果有定时任务，启动beat -4 等待任务被worker执行 -5 在视图函数中，查询任务执行的结果 1、模拟写一个异步秒杀任务
后端
view.py
from celery.result import AsyncResult from celery_task.celery import app from celery_task.task import sckill_task # 秒杀接口 class SeckillView(ViewSet): # 开启秒杀 @action(methods=['GET'], detail=False) def seckill(self, request): # 获取商品链接 goods_id = request.query_params.get('goods_id') # 将任务提交给worker res = sckill_task.delay(goods_id) # 将任务的ID反馈给前端 return APIResponse(task_id=str(res)) # 查询秒杀结果 @action(methods=['GET'], detail=False) def get_result(self, request): # 前端将任务ID产过来，用于接收结果 task_id = request.query_params.get('task_id') # 调用接口，查询结果 a = AsyncResult(id=task_id, app=app) if a.successful(): result = a.get() if result: return APIResponse(msg='秒杀成功') else: return APIResponse(code=101, msg='手速满了，秒杀失败') elif a.status == 'PENDING': return APIResponse(code=666, msg='加速秒杀中') return APIResponse(msg='错误') celery.py ---->秒杀任务
import random # 秒杀函数 @app.task() def sckill_task(goods_id): print('商品正在秒杀中') time.sleep(random.choice([6, 7, 8, 9])) print('商品秒杀结束') return random.choice([True, False]) 前端：
<template> <div> <button @click="clickHandle">点击秒杀</button> </div> </template> <script> export default { name: "Template", data() { return { // 用于接收任务ID task_id: '', // 用户存放定时任务 t: '' } }, methods: { // 用户点击秒杀后发送请求 clickHandle() { // 向厚点提交秒杀任务 this.$axios.get(this.$settings.BASE_URL + '/user/seckill/seckill/?goods_id=1').then(res => { // 判断任务是否提交成功 if (res.data.code == 100) { // 提交成功会获取到任务ID this.task_id = res.data.task_id // 告知用户商品正在秒杀中 this.$message('正在秒杀中') // 启动一个定时任务，每隔3秒向后端发送请求，获取任务是否提交成功 this.t = setInterval(res => { // 定时向后端发送请求，判断秒杀结果 this.$axios.get(this.$settings.BASE_URL + '/user/seckill/get_result/?task_id=' + this.task_id).then(res => { // 判断任务是否结束 if (res.data.code == 666) { this.$message(res.data.msg) // 任务结束反馈结果，关闭定时器 } else { this.$message(res.data.msg) // 关闭定时器 clearInterval(this.t) this.t = '' } }) }, 3000) } }) } } } </script> 2、总结
- 第一步：将celery包复制到项目路径下
-luffy_api -celery_task #celery的包路径 celery.py # 一定不要忘了一句话 import os # 重点：celery中使用djagno,任务中可能会使用django的orm，缓存，表模型。。。。一定要加 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffy_api.settings.dev') -luffy_api #源代码路径 - 第二步：在需要使用异步的地方导入celery实例即可使用
-视图函数中使用，导入任务 -任务.delay() # 提交任务 -
第三步：启动worker，如果有定时任务，启动beat
-
第四步： 等待任务被worker执行
-
第五步：在视图函数中，查询任务结果
