python celery beat实现定时任务的示例代码

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
"""

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

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)
    },
}