python celery beat实现定时任务

在Celery在python中的应用除了实现异步任务（async task)外也可以执行定时任务(beat)

1.Celery定时任务是什么？

Celery默认任务单元由任务生产者触发,但有时可能需要其自动触发, 而beat进程正是负责此类任务,能够自动触发定时/周期性任务.

在这里插入图片描述

只需要在配置中配置好周期任务，然后在运行一个周期任务触发器(beat)即可

2.直接上代码

目录结构如下：

在这里插入图片描述

celery_app.py 文件代码如下：

import os
import sys
import time
import celery
from pathlib import Path
from datetime import timedelta

# 实例化celery对象
app = celery.Celery(
    "celery_worker",
    backend="redis://:@127.0.0.1:6379/4",
    broker="redis://:@127.0.0.1:6379/5",
    include=[
        "celery_worker.email.tasks"
    ],
)

# celery beat 定时任务
beat_schedule = {
    'periodic_task-every-minute': {
        # 'task': 'celery_worker.email.tasks.add',
        'task': 'chain.send_chains',
        'schedule': timedelta(seconds=10),
        'args': (11, 22)
    },
}

# 配置文件
app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_default_queue="normal",
    timezone="Asia/Shanghai",
    enable_utc=False,
    task_ignore_result=True,
    redis_max_connections=100,
    result_expires=3600,
    beat_schedule=beat_schedule
)

"""
celery -A celery_worker.celery_app worker -l info
celery -A celery_worker.celery_app beat
"""

email.tasks.py 代码如下：

from loguru import logger
# 模块化之后
from celery_worker.celery_app import app

@app.task(name='chain.send_chains')
def add(x, y):
    logger.info(f'number_add 进来了...x:{x}, y:{y}')
    return x + y

然后顺序启动 worker 和 beat 定时任务（记得两个都必须启动）

执行如下命令：

celery -A celery_worker.celery_app worker -l info   （启动干活的人）

celery -A celery_worker.celery_app beat  （启动定时任务 类似crontab）

效果如下：

在这里插入图片描述

其实简单的来说就是这点代码：

beat_schedule = {
    'periodic_task-every-minute': {
        # 'task': 'celery_worker.email.tasks.add',
        'task': 'chain.send_chains',
        'schedule': timedelta(seconds=10),
        'args': (11, 22)
    },
}