Celery框架

1. 什么是celery

celery是一个简答, 灵活且可靠的, 处理大量消息的分布式系统, 专注于实时处理的异步任务队列, 同时也支持任务调度.

这里面提到了一个概念: 分布式系统

一个系统应用(网站), 会有相关组件(web服务器, web应用, 数据库, 消息中间件), 将它们架构在不同的服务器上, 不同的服务器的不同组件之间通过消息通信的方式来实现协调工作, 这种模式叫分布式系统
优点: 实现负载均衡, 避免单点故障

Celery的架构存在三部分组成, 消息中间件(message broker), 任务执行单元(worker)和任务结果储存(task result store)组成.

消息中间件

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

任务执行单元

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

任务结果储存

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

celery具有以下优点

Simple(简单)
Celery 使用和维护都非常简单，并且不需要配置文件。

Highly Available（高可用）
woker和client会在网络连接丢失或者失败时，自动进行重试。并且有的brokers 也支持“双主”或者“主／从”的方式实现高可用。

Fast（快速）
单个的Celery进程每分钟可以处理百万级的任务，并且只需要毫秒级的往返延迟（使用 RabbitMQ, librabbitmq, 和优化设置时）

Flexible（灵活）
Celery几乎每个部分都可以扩展使用，自定义池实现、序列化、压缩方案、日志记录、调度器、消费者、生产者、broker传输等等。

另外， Celery还支持不同的并发和序列化的手段

并发：Prefork, Eventlet, gevent, threads/single threaded
序列化：pickle, json, yaml, msgpack. zlib, bzip2 compression， Cryptographic message signing 等等

2. celery使用场景

celery是一个强大的分布式任务队列的异步处理框架，它可以让任务的执行完全脱离主程序，甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务（async task）和定时任务（crontab)。

异步任务：将耗时操作任务提交给Celery去异步执行，比如发送短信/邮件、消息推送、音视频处理等等

定时任务：定时执行某件事情，比如每天数据统计

3. Celery执行异步任务

执行流程:

任务定义：首先，需要定义 Celery 任务。任务可以是 Python 函数或类的方法，用 @task 装饰器进行修饰。任务定义了要执行的具体操作。
任务发布：任务的发布者（生产者）使用 Celery 提供的客户端接口，将任务发送到任务队列中（例如 RabbitMQ 或 Redis）
任务调度：任务队列中的任务被 Celery 的worker（消费者）获取并执行。worker可以在同一台或多台机器上运行，构成一个分布式任务执行的集群。
任务执行：一旦worker获取到任务，它会根据任务的定义，调用相应的函数或方法执行具体的操作。任务可以是同步的或异步的，取决于任务定义中是否使用了异步执行的修饰器。
任务结果：任务执行完成后，worker将执行结果返回给消息代理，并将结果存储在结果后端（例如数据库、缓存或消息队列）中，以供后续查询。

3.1 基本使用

celery_task.py

# -*- encoding:utf-8 -*-
# @time: 2023/5/25 17:06
# @author: ifeng
import celery
import time


backend = 'redis://:123456@127.0.0.1:6379/1'  # 配置任务结果储存位置
broker = 'redis://:123456@127.0.0.1:6379/2'  # 配置消息中间件

cel = celery.Celery('test', backend=backend, broker=broker)


@cel.task
def send_email(name):
    print('向%s发送邮件' % name)
    time.sleep(5)
    print('向%s发送邮件完成' % name)
    return '%s, 发送完毕' % name


@cel.task
def send_msg(name):
    print('向%s发送信息' % name)
    time.sleep(5)
    print('向%s发送信息完毕' % name)
    return '%s, 发送完毕' % name

produce_task.py

# -*- encoding:utf-8 -*-
# @time: 2023/5/25 17:21
# @author: ifeng

from celery_task import send_email, send_msg


result = send_email.delay('ifeng')  # send_task
print(result.id)
# print(result.get())  # 获取返回值

result = send_msg.delay('ifeng')
print(result.id)

result.py

# -*- encoding:utf-8 -*-
# @time: 2023/5/25 20:04
# @author: ifeng
from celery.result import AsyncResult
from celery_task import cel


async_result = AsyncResult(id='75ff05e6-912a-44f4-a481-eb5bff411488', app=cel)

if async_result.successful():
    result = async_result.get()
    print(result)
    # result.forget() # 将结果删除
elif async_result.failed():
    print('执行失败')
elif async_result.status == 'PENDING':
    print('任务等待中被执行')
elif async_result.status == 'RETRY':
    print('任务异常后正在重试')
elif async_result.status == 'STARTED':
    print('任务已经开始被执行')

开启work: celery worker -A celery_task -l info -P eventlet

3.2 多任务结构

注意创建文件夹的时候要创建python package, 否则会出现导报不成功

celery.py

# -*- encoding:utf-8 -*-
# @time: 2023/5/25 20:24
# @author: ifeng
from celery import Celery


cel = Celery('celery_demo',
             broker='redis://:123456@127.0.0.1:6379/1',
             backend='redis://:123456@127.0.0.1:6379/2',
             # 包含以下两个任务文件，去相应的py文件中找任务，对多个任务做分类
             include=['celery_task.task01',
                      'celery_task.task02'
                      ])

# 时区
cel.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
cel.conf.enable_utc = False

task01.py

# -*- encoding:utf-8 -*-
# @time: 2023/5/25 20:25
# @author: ifeng
import time
from .celery import cel


@cel.task
def send_email(res):
    time.sleep(5)
    return "完成向%s发送邮件任务" % res

task02.py

# -*- encoding:utf-8 -*-
# @time: 2023/5/25 20:25
# @author: ifeng
import time

from .celery import cel


@cel.task
def send_msg(name):
    time.sleep(5)
    return "完成向%s发送短信任务" % name

produce_task.py

# -*- encoding:utf-8 -*-
# @time: 2023/5/25 20:28
# @author: ifeng

from celery_task.task01 import send_email
from celery_task.task02 import send_msg


result = send_email.delay("ifeng")
print(result.id)

result = send_msg.delay('ifeng')
print(result.id)

result.py

# -*- encoding:utf-8 -*-
# @time: 2023/5/25 20:30
# @author: ifeng
from celery.result import AsyncResult

from celery_task.celery import cel

async_result = AsyncResult(id="90e941e4-97e0-4c53-8609-91fb8a91768e", app=cel)

if async_result.successful():
    result = async_result.get()
    print(result)
    # result.forget() # 将结果删除,执行完成，结果不会自动删除
    # async.revoke(terminate=True)  # 无论现在是什么时候，都要终止
    # async.revoke(terminate=False) # 如果任务还没有开始执行呢，那么就可以终止。
elif async_result.failed():
    print('执行失败')
elif async_result.status == 'PENDING':
    print('任务等待中被执行')
elif async_result.status == 'RETRY':
    print('任务异常后正在重试')
elif async_result.status == 'STARTED':
    print('任务已经开始被执行')

开启work：celery worker -A celery_task -l info -P eventlet，添加任务（执行produce_task.py)，检查任务执行结果（执行check_result.py）

4.celery执行定时任务

4.1 简单结构实现定时任务

produce_task.py

from celery_task import send_email
from datetime import datetime

# 方式一
# v1 = datetime(2020, 3, 11, 16, 19, 00)
# print(v1)
# v2 = datetime.utcfromtimestamp(v1.timestamp())
# print(v2)
# result = send_email.apply_async(args=["egon",], eta=v2)
# print(result.id)

# 方式二
ctime = datetime.now()
# 默认用utc时间
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
from datetime import timedelta
time_delay = timedelta(seconds=10)
task_time = utc_ctime + time_delay

# 使用apply_async并设定时间
result = send_email.apply_async(args=["egon"], eta=task_time)
print(result.id)

4.2 多目录结构实现定时任务

celery.py

from datetime import timedelta
from celery import Celery
from celery.schedules import crontab

cel = Celery('tasks', broker='redis://127.0.0.1:6379/1', backend='redis://127.0.0.1:6379/2', include=[
    'celery_tasks.task01',
    'celery_tasks.task02',
])
cel.conf.timezone = 'Asia/Shanghai'
cel.conf.enable_utc = False

cel.conf.beat_schedule = {
    # 名字随意命名
    'add-every-10-seconds': {
        # 执行tasks1下的test_celery函数
        'task': 'celery_tasks.task01.send_email',
        # 每隔2秒执行一次
        # 'schedule': 1.0,
        # 'schedule': crontab(minute="*/1"),
        'schedule': timedelta(seconds=6),  # 间隔时间
        # 传递参数
        'args': ('张三',)  # 相关参数
    },
    # 'add-every-12-seconds': {
    #     'task': 'celery_tasks.task01.send_email',
    #     每年4月11号，8点42分执行
    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),
    #     'args': ('张三',)
    # },
}

# 启动 Beat 程序$ celery beat -A proj<br># Celery Beat进程会读取配置文件的内容，周期性的将配置中到期需要执行的任务发送给任务队列
# 如果这个没有关会导致历史遗留问题, 导致储存多个任务在redis的List中, 这个需要关注

# 之后启动 worker 进程.$ celery -A proj worker -l info 或者$ celery -B -A proj worker -l info

5.django中使用celery

项目根目录创建celery包，目录结构如下：

mycelery/
├── config.py
├── __init__.py
├── main.py
└── sms/
    ├── __init__.py
    ├── tasks.py

配置文件config.py:

broker_url = 'redis://127.0.0.1:6379/15'
result_backend = 'redis://127.0.0.1:6379/14'

任务文件tasks.py：

# celery的任务必须写在tasks.py的文件中，别的文件名称不识别!!!
from mycelerys.main import app
import time


import logging
log = logging.getLogger("django")

@app.task  # name表示设置任务的名称，如果不填写，则默认使用函数名做为任务名
def send_sms(mobile):
    """发送短信"""
    print("向手机号%s发送短信成功!"%mobile)
    time.sleep(5)

    return "send_sms OK"

@app.task  # name表示设置任务的名称，如果不填写，则默认使用函数名做为任务名
def send_sms2(mobile):
    print("向手机号%s发送短信成功!" % mobile)
    time.sleep(5)

    return "send_sms2 OK"

最后在main.py主程序中对django的配置文件进行加载

# 主程序
import os
from celery import Celery
# 创建celery实例对象
app = Celery("sms")

# 把celery和django进行组合，识别和加载django的配置文件
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celeryPros.settings.dev')

# 通过app对象加载配置
app.config_from_object("mycelerys.config")

# 加载任务
# 参数必须必须是一个列表，里面的每一个任务都是任务的路径名称
# app.autodiscover_tasks(["任务1","任务2"])
app.autodiscover_tasks(["mycelerys.sms",])

# 启动Celery的命令
# 强烈建议切换目录到mycelery根目录下启动
# celery -A mycelery.main worker --loglevel=info

Django视图调用：

from django.shortcuts import render

# Create your views here.


from django.shortcuts import render,HttpResponse
from mycelerys.sms.tasks import send_sms,send_sms2
from datetime import timedelta

from datetime import datetime
def test(request):

    ################################# 异步任务

    # 1. 声明一个和celery一模一样的任务函数，但是我们可以导包来解决

    # send_sms.delay("110")
    # send_sms2.delay("119")
    # send_sms.delay() 如果调用的任务函数没有参数，则不需要填写任何内容


    ################################# 定时任务

    # ctime = datetime.now()
    # # 默认用utc时间
    # utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
    # time_delay = timedelta(seconds=10)
    # task_time = utc_ctime + time_delay
    # result = send_sms.apply_async(["911", ], eta=task_time)
    # print(result.id)

    return HttpResponse('ok')

6.Celery作为第三方模块集成到项目中

首先看一下目录结构

luffycityapi/           # 服务端项目根目录
└── luffycityapi/       # 主应用目录
    ├── apps/           # 子应用存储目录  
    ├   └── users/            # django的子应用
    ├       └── tasks.py      # [新增]分散在各个子应用下的异步任务模块
    ├── settings/     # [修改]django的配置文件存储目录[celery的配置信息填写在django配置中即可]
    ├── __init__.py   # [修改]设置当前包目录下允许外界调用celery应用实例对象
    └── celery.py     # [新增]celery入口程序，相当于上一种用法的main.py

6.1 具体执行流程

创建celery.py文件, 定义celery框架, 会帮集成监听消息功能
在dev.py配置文件中新增相关配置信息
在app中创建任务tasks
在app的视图函数的api向队列(redis,rabbitmq)中推送信息
消费者从队列中获取消息并扔给worker, worker可以在不同服务器上, 构成一个分布式集群

6.2 代码部分

luffycityapi/celery.py, 主应用目录下创建celery入口程序, 创建celery对象并加载配置和异步任务

import os
from celery import Celery

# 必须在实例化celery应用对象之前执行
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffycityapi.settings.dev')

# 实例化celery对象
app = Celery('luffycityapi')

# 指定任务的队列名称
app.conf.task_default_queue = 'Celery'

# 也可以把配置写在django的项目配置中
app.config_from_object('django.conf:settings', namespace='CELERY')  # 设置django中配置信息以 "CELERY_"开头为celery的配置信息

# 自动根据配置查找django的所有子应用下的tasks任务文件
app.autodiscover_tasks()

settings/dev.py, 新增celery配置信息, 代码

# Celery异步任务队列框架的配置项[注意：django的配置项必须大写，所以这里的所有配置项必须全部大写]
# 任务队列
CELERY_BROKER_URL = 'redis://@127.0.0.1:6379/14'
# 结果队列
CELERY_RESULT_BACKEND = 'redis://@127.0.0.1:6379/15'
# 时区，与django的时区同步
CELERY_TIMEZONE = TIME_ZONE
# 防止死锁
CELERY_FORCE_EXECV = True
# 设置并发的worker数量
CELERYD_CONCURRENCY = 200
# 设置失败允许重试[这个慎用，如果失败任务无法再次执行成功，会产生指数级别的失败记录]
CELERY_ACKS_LATE = True
# 每个worker工作进程最多执行500个任务被销毁，可以防止内存泄漏，500是举例，根据自己的服务器的性能可以调整数值
CELERYD_MAX_TASKS_PER_CHILD = 500
# 单个任务的最大运行时间，超时会被杀死[慎用，有大文件操作、长时间上传、下载任务时，需要关闭这个选项，或者设置更长时间]
CELERYD_TIME_LIMIT = 10 * 60
# 任务发出后，经过一段时间还未收到acknowledge, 就将任务重新交给其他worker执行
CELERY_DISABLE_RATE_LIMITS = True
# celery的任务结果内容格式
CELERY_ACCEPT_CONTENT = ['json', 'pickle']

# 之前定时任务（定时一次调用），使用了apply_async({}, countdown=30);
# 设置定时任务（定时多次调用）的调用列表，需要单独运行SCHEDULE命令才能让celery执行定时任务：celery -A mycelery.main beat，当然worker还是要启动的
# https://docs.celeryproject.org/en/stable/userguide/periodic-tasks.html
from celery.schedules import crontab
CELERY_BEAT_SCHEDULE = {
    "user-add": {  # 定时任务的注册标记符[必须唯一的]
        "task": "delayed_task",   # 定时任务的任务名称
        "schedule": 10,  # 定时任务的调用时间，10表示每隔10秒调用一次add任务
        # "schedule": crontab(hour=7, minute=30, day_of_week=1),,  # 定时任务的调用时间，每周一早上7点30分调用一次add任务
    }
}

luffycityapi/__init__.py, 主应用下初始化,

import pymysql
from .celery import app as celery_app

pymysql.install_as_MySQLdb()

__all__ = ['celery_app']

users/tasks.py, 创建任务列表

from celery import shared_task
from luffycityapi.utils.ronglianyunapi import sms_send

# 记录日志
import logging
logging.getLogger('django')


@shared_task(name="send_sms")
def send_sms(tid, mobile, datas):
    try:
        return sms_send(tid, mobile, datas)
    except Exception as e:
        logging.error(f"手机号：{mobile}，发送短信失败错误: {e}")


@shared_task(name="delayed_task")
def delayed_task():
    logging.info(f'延时消息发送成功啦！！！')
    return "delayed_task 执行完毕"

此时在django中用户发送短信, 就可以改成异步发送短信了

user/views.py, 视图中调用异步发送短信的任务

from .tasks import send_sms
send_sms.delay(settings.RONGLIANYUN.get("reg_tid"),mobile, datas=(code, time // 60))

6.3 启动celery

cd ~/Desktop/luffycity/luffycityapi
# 1. 普通运行模式，关闭终端以后，celery就会停止运行
celery -A luffycityapi worker  -l INFO

# 2. 启动多worker进程模式，以守护进程的方式运行，不需要在意终端。但是这种运行模型，一旦停止，需要手动启动。
celery multi start worker -A luffycityapi -E --pidfile="/home/ifeng/Desktop/luffycity/luffycityapi/logs/worker1.pid" --logfile="/home/ifeng/Desktop/luffycity/luffycityapi/logs/celery.log" -l info -n worker1

# 3. 启动多worker进程模式
celery multi stop worker -A luffycityapi --pidfile="/home/ifeng/Desktop/luffycity/luffycityapi/logs/worker1.pid"

6.4 补充

也可以在django终端下调用celery

$ python manage.py shell
>>> from users.tasks import delayed_task
>>> res = delayed_task.delay()
>>> res = delayed_task.apply_async(countdown=15)
>>> res.id
'893c31ab-e32f-44ee-a321-8b07e9483063'
>>> res.state
'SUCCESS'
>>> res.result

定时任务的调用器启动, 可以在运行了worker以后, 使用一下命令:

cd ~/Desktop/luffycity/luffycityapi
celery -A luffycityapi beat

定时任务执行结果展示(十秒执行一次):

[2023-06-02 21:17:18,144: INFO/ForkPoolWorker-3] 延时消息发送成功啦！！！
[2023-06-02 21:17:18,147: INFO/ForkPoolWorker-3] Task delayed_task[3a1033fe-b1a2-4f7f-9fe6-61d1ef9816ab] succeeded in 0.0034034299997074413s: 'delayed_task 执行完毕'
[2023-06-02 21:17:28,140: INFO/MainProcess] Task delayed_task[905f3644-297a-430a-ae23-75be7e3a2ad3] received
[2023-06-02 21:17:28,142: INFO/ForkPoolWorker-3] 延时消息发送成功啦！！！
[2023-06-02 21:17:28,143: INFO/ForkPoolWorker-3] Task delayed_task[905f3644-297a-430a-ae23-75be7e3a2ad3] succeeded in 0.0014366229997904156s: 'delayed_task 执行完毕'
[2023-06-02 21:17:38,140: INFO/MainProcess] Task delayed_task[c1962489-e615-4e6d-a5b0-171826d1b56e] received
[2023-06-02 21:17:38,141: INFO/ForkPoolWorker-3] 延时消息发送成功啦！！！
[2023-06-02 21:17:38,143: INFO/ForkPoolWorker-3] Task delayed_task[c1962489-e615-4e6d-a5b0-171826d1b56e] succeeded in 0.0013498770003934624s: 'delayed_task 执行完毕'