使用celery创建定时任务

测试环境：
ubuntu:20.04
python:3.6
django:2.2
celery:4.4.7

创建django项目

添加celery的配置文件，目录结构如下图

修改配置文件

这里我使用redis作为消息中间件，任务的信息和结果返回都存储在redis中。
/proj/settings.py文件

CELERY_BROKER_URL = 'redis://localhost:6379/0' # 工人信息，以及任务信息
CELERY_ACCEPT_CONTENT = ['json']
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1' # 任务结果存储
CELERY_TASK_SERIALIZER = 'json'
CELERY_TIME_ZONE = 'Asia/Shanghai'
CELERY_ENABLE_UTC = True

/proj/celery.py文件

import os
from celery import Celery
# 获取celery在settings中的配置信息
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')
# celery实例对象创建
app = Celery('proj')
# 使用配置信息，其中namespace作用为所有的celery配置都需要使用CELERY作为前缀，且最好所有字母都大写
app.config_from_object('django.conf:settings', namespace='CELERY')
# 获取所有的task,官方文档中说可以自动获取所有的task，但是我测试时需要加上task否则他会显示为注册类型
app.autodiscover_tasks(['demoapp.tasks.mul'])

# 下面的设置就是关于调度器beat的设置
app.conf.beat_schedule = {
    'test': {  # 取个名字
        'task': 'demoapp.tasks.mul',  # 设置是要将哪个任务进行定时
        'schedule': crontab(hour=23, minute=59, day_of_week=[0,1,2,3,4,5,6]),  # 调用crontab进行具体时间的定义.如果想以秒为单位做定时可以'schedule': 10表示每10秒运行该任务
        'args':(2,3) # 参数
    }
}

@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

测试代码

/demoapp/task.py

from celery import shared_task

@shared_task
def mul(x, y):
    return x * y

运行命令

在项目目录下运行此命令

celery -A proj worker -B -l info

其中-A表示创建的celery对象，-B表示启动定时，-l info日志输出等级是info


参考官方文档

定时任务crontab用法