Celery - 分布式任务队列

1 celery简介

1.1 什么是celery

Celery是一个用Python编写的分布式任务队列框架。它允许开发者将任务分发到多台服务器或进程中执行，从而实现高效的异步任务处理。

Celery框架基于消息传递实现，任务可以在多个节点之间异步传递和执行，这大大提高了任务的可靠性和可伸缩性。Celery还支持多种消息传递中间件，如RabbitMQ、Redis、Amazon SQS等。

• 注意：

1）celery可以不依赖任何服务器，通过自身命令，启动服务(内部支持socket)

2）celery服务为为其他项目服务提供异步解决任务需求的

注：会有两个服务同时运行，一个是项目服务，一个是celery服务，项目服务将需要异步处理的任务交给celery服务，celery就会在需要时异步完成项目的需求

在django中想用异步，交个celery来做，其他不需要的则django自己来做

celery和django是两个项目服务

1.2 celery架构

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

（1）消息中间件 message broker

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

（2）任务执行单元 worker

Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。

（3）任务结果存储 task result store

Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, redis等

任务中间件 broker:其他项目服务提交的异步任务，放在里面排队 【需要借助第三方：redis，rabbitmq】

任务执行单元 worker:真正执行异步任务的进程

结果存储 backend：结果存储，函数的返回结果存到backend中，【需要借助第三方：redis，mysql】

（4）使用场景

异步执行：解决耗时任务

延迟执行：解决延迟任务

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

2 Celery安装与使用

2.1 安装

pip install celery

消息中间件：RabbitMQ/Redis

windows系统上要安装这个模块

pip install eventlet

2.2 快速使用

① 第1步:创建celery app与创建任务

• celery_test.py中

from celery import Celery

# 先配置好
# 消息中间件broker：提交的异步任务，放在broker中
broker = 'redis://127.0.0.1:6379/3'
# 任务结果存储backend：执行完的结果，放在backend中
backend = 'redis://127.0.0.1:6379/4'

# 第1步：类实例化得到对象

celery_app = Celery('test1', broker=broker, backend=backend)


# 第一个参数是名字：可以用__name__
# 第二个、第三个参数是：broker、backend

# 第2步：写任务，用装饰器的形式
@celery_app.task
def add_test(a, b):
    import time
    time.sleep(3)
    print(a + b)
    return a + b

# 第3步：提交任务，在别的程序中写提交任务

② 第2步:提交任务：在别的程序中

• django - task.py 中

# 在这里编写celery的提交任务


from celery_test import add_test

# 同步调用
# res = add_test(5,6)
# print(res)

# 异步调用

res = add_test.delay(1, 3)
print(res)
# 这时候，任务还没执行，说明celery的worker还没执行，只是提交上去了
# 422c683a-a255-4c7c-be27-ccb76e7ee19f


# 启动 worker

③ 第3步:启动celery(app)服务 【也就是worker】

• 非windows

4.x之前版本
celery worker -A celery的包名 -l info

5.x之后的版本
celery -A celery的包名 worker -l info

• windows

pip3 install eventlet

4.x之前版本
celery worker -A celery的包名 -l info -P eventlet

5.x之后的版本
celery -A celery的包名 worker -l info -P eventlet

④ 第4步:worker会执行消息中间件中的任务，并把结果存起来

worker执行完会将结果存储在redis中

⑤ 第5步:查看结果【通过程序获得结果】

• django - get_result.py中

# 查看celery的worker结果
from celery_test import celery_app

from celery.result import AsyncResult

id = '17e170a4-0c74-455d-afa7-4c5c893de152'
if __name__ == '__main__':
    async_obj = AsyncResult(id=id, app=celery_app)
    if async_obj.successful():
        result = async_obj.get()
        print(result)
    elif async_obj.failed():
        print('任务失败')
    elif async_obj.status == 'PENDING':
        print('任务等待中被执行')
    elif async_obj.status == 'RETRY':
        print('任务异常后正在重试')
    elif async_obj.status == 'STARTED':
        print('任务已经开始被执行')

3 celery包结构 - 在项目中使用celery

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

3.1 在项目中使用celery操作步骤：

① 第1步：新建包 - 编写celery项目：创建、注册、配置

在celery项目的包结构中

• celery.py中，注册任务和配置

from celery import Celery

# 先配置好
# 消息中间件broker：提交的异步任务，放在broker中
broker = 'redis://127.0.0.1:6379/3'
# 任务结果存储backend：执行完的结果，放在backend中
backend = 'redis://127.0.0.1:6379/4'

# 第1步：类实例化得到对象
# 不要忘记include
celery_app = Celery('test1', broker=broker, backend=backend,
                    include=['celery_project.order_task', 'celery_project.user_task'])

# 新建任务py文件

② 第2步：在包内部task.py中，编写异步任务

# 发送短信任务


from .celery import celery_app
import time


@celery_app.task
def send_sms(mobile, code):
    print("发送%s短信成功：验证码为%s" % (mobile, code))
    time.sleep(2)
    return True

③ 第3步：在celery项目的路径外，启动worker

/Luffy/luffy_api/script   celery -A celery_test worker -l info

④ 第4步：在其他项目中，提交任务

被提交到中间件中，等待worker执行，因为worker启动了，就会被worker执行

"提交celery任务"


from celery_project.user_task import send_sms


# 同步调用
# res = send_sms('13900000000','1234')
# print(res) # 发送13900000000短信成功：验证码为1234


# 异步调用
task_id = send_sms.delay('13900000000','1234')
print(task_id)
# 8d1aea99-1ba0-4248-a91f-786ac9ffa1c9

⑤ 第5步：worker执行完，结果存到backend

这里使用的是redis

⑥ 第6步：在其他项目中，查看结果

# 查看celery的worker结果
from celery_project.celery import celery_app

from celery.result import AsyncResult

id = '8d1aea99-1ba0-4248-a91f-786ac9ffa1c9'
if __name__ == '__main__':
    async_obj = AsyncResult(id=id, app=celery_app)
    if async_obj.successful():
        result = async_obj.get()
        print(result)
    elif async_obj.failed():
        print('任务失败')
    elif async_obj.status == 'PENDING':
        print('任务等待中被执行')
    elif async_obj.status == 'RETRY':
        print('任务异常后正在重试')
    elif async_obj.status == 'STARTED':
        print('任务已经开始被执行')

4 异步任务、延迟任务、定时任务

4.1 异步任务

任务名.delay(args=[参数])

4.2 延迟任务

app.apply_async(args=[参数],eta=时间对象)

• 演示

# 延时任务：延迟20秒
from datetime import datetime, timedelta

eta = datetime.utcnow() + timedelta(seconds=20)
# celery默认配置文件中使用的utc的时间
res2 = send_sms.apply_async(args=['222222', '8888'], eta=eta)

4.3 定时任务

每间隔一段时间，执行某个任务

定时任务需要启动beat

beat:【定时提交任务的进程：配置在app.conf.beat_schedule 中的任务】

worker:执行任务

① 第一步:任务的定时配置

配置中可以配置多个

from celery.schedules import crontab
# 新建任务py文件
celery_app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
celery_app.conf.enable_utc = False
# celery的配置文件#####
# 任务的定时配置
celery_app.conf.beat_schedule = {
    'send_sms': {
        'task': 'celery_project.user_task.send_sms',  # 导包，一定从celery包开始导入
        # 'schedule': timedelta(seconds=3),  # 时间对象
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八点
        'schedule': crontab(hour=23, minute=55),  # 每天9点43
        'args': ('18888888', '6666'),
    },
}

② 第二步:启动beat

beat 是定时提交任务的进程，执行配置在app.conf.beat_schedule中的任务

celery -A celery_project beat -l info

③ 第三步:启动worker

celery -A celery_project worker -l info 

win:celery -A celery_task worker -l info -P eventlet

注意：

• 启动命令的执行位置，如果是包结构一定要在包这层，也就是创建了celery包的上一层

• include=['xxx']，路径是从包名下开始导的

5 django中使用celery

使用定时任务，除了celery还可以使用别的：APSchedule第三方模块执行定时任务

5.1 使用步骤

① 第一步：将写好的包放在项目路径下

-luffy_api
    -celery_task #celery的包路径
       -luffy_api  #源代码路径

② 第二步：在使用提交一步任务的位置，导入使用即可

视图函数中使用，导入任务

任务.delay()  # 提交任务

③ 第三步：启动worker，如果有定时任务，启动beat

celery -A celery包 worker -l info

④ 第四步：等待任务被worker执行

⑤ 在视图函数中，查询任务执行的结果

5.2 案例：秒杀功能

异步操作做秒杀，提高并发量

① 前端逻辑：

1） 秒杀按钮，用户点击 发送ajax请求到后端

② 后端逻辑：

2） 视图函数---》提交秒杀任务---》借助于celery，提交到中间件中了

3） 当次秒杀的请求，就回去了，携带者任务id号在前端

4） 前端开启定时任务，每隔3s钟，带着任务，向后端发送请求，查看是否秒杀成功

5） 后端还可能出现的情况

​ I) 任务还在等待被执行----》返回给前端，前端继续每隔3s发送一次请求
​ II) 任务执行完了，秒杀成功了---》返回给前端，恭喜您秒杀成功--》关闭前端定时器
​ III) 任务执行完了，秒杀失败了---》返回给前端，秒杀失败--》关闭前端定时器

本栏目推荐文章
• 分布式限流——基于Redis的Lua脚本限流实现
• 第二章 Spring Boot 整合 Kafka消息队列 生产者
• 分布式代码执行分析
• 毕设具体任务
• 简易机器学习笔记（十一）opencv 简易使用-人脸识别、分类任务
• .net core分布式定时任务调度系统
• 消息队列面试题：为什么要使用消息队列？
• Kettle设置定时跑任务
• 分布式缓存的实现和使用
• 开源：Taurus.DTC 微服务分布式事务框架，支持 .Net 和 .Net Core 双系列版本

队列 分布式 任务 Celery队列 分布式 任务celery 队列 任务celery worker 分布式 框架 任务celery 队列 分布式 任务 消息 任务celery 课程celery接口 任务 django-celery任务django celery eventlet任务django celery django-celery任务 数据库 数据 周期性 周期 任务celery