112 - 第三方库 - Celery - 定时分布式任务执行

celery库的简介

Python 的 Celery 是一个强大的分布式任务队列系统，用于异步执行耗时的任务。它支持多种消息传递系统，如 RabbitMQ、Redis 等，用于任务分发和结果存储。

Celery 提供了易于使用的 API，允许开发者以声明式方式定义任务，支持定时任务、周期性任务以及基于时间的复杂任务调度。它还具备高可用性和可扩展性，非常适合构建大规模的分布式系统。

Celery 广泛应用于 Web 应用、数据处理、定期报告生成等场景，可以显著提高应用程序的响应速度和处理能力，是实现后台任务处理和异步操作的优选工具。

安装定时任务库 celery 命令、任务监控 flower 命令

# celery 安装
pip install celery

可以看见不仅仅安装了celery，还安装了amqp-5.2.0 billiard-4.2.1 celery-5.4.0 click-8.1.7 click-didyoumean-0.3.1 click-plugins-1.1.1 click-repl-0.3.0 kombu-5.4.2 vine-5.1.0

# celery 监控 flower
pip install flower

目录

project
.
├── config.py
├── __init__.py
├── period_task.py
└── tasks.py

celery 实例初始化

__init__.py

from celery import Celery
app = Celery('project')  # 创建 Celery 实例
app.config_from_object('project.config')

配置 project.config

config.py

BROKER_URL = 'redis://10.8.238.2:6379/0' # Broker配置，使用Redis作为消息中间件
CELERY_RESULT_BACKEND = 'redis://10.8.238.2:6379/0' # BACKEND配置，这里使用redis
CELERY_RESULT_SERIALIZER = 'json' # 结果序列化方案
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # 任务过期时间
CELERY_TIMEZONE='Asia/Shanghai'   # 时区配置
CELERY_IMPORTS = (     # 指定导入的任务模块,可以指定多个
    'project.tasks',
    'project.period_task'
)

定义任务

tasks.py

import celery
import time
from celery.utils.log import get_task_logger
from project import app
@app.task
def sum(x, y):
    return x + y
@app.task
def mul(x, y):
    time.sleep(5)
    return x * y

https://mp.weixin.qq.com/s?__biz=MzAxMjUyNDQ5OA==&mid=2653562766&idx=1&sn=86bd7e0a834722d7554c4a1341162a69&chksm=806e0933b7198025fd60ec95973ea187450ce6a8399f4891aad13fec951b24106cf95a85e21c&scene=27