django-celery-beat搭建定时任务的实现

下面我将详细讲解如何使用Django与Celery Beat搭建定时任务的实现。

这里简单介绍一下Django和Celery

Django

Django是一个由Python编写的免费、开源的Web框架。它遵循了MVC设计模式，且拥有灵活的URL路由和完善的后台管理，使得Django能够非常方便地开发高质量的Web应用程序。

Celery

Celery是一个由Python编写的分布式任务队列，它允许您异步地执行任务、周期性地执行任务以及管理任务等。

Celery Beat

Celery Beat是Celery中的一个组件，它是用来完成调度任务的，也就是实现定时任务的核心组件。

搭建定时任务的实现

下面我们就开始讲解如何使用Django和Celery Beat搭建定时任务的实现。

第一步：创建Django项目和应用

使用Django创建一个项目及应用：

$ django-admin startproject project_name
$ python manage.py startapp app_name

第二步：安装Celery和django-celery-beat

在创建好的应用中，通过pip安装以下两个库：

$ pip install celery
$ pip install django-celery-beat

第三步：配置Celery和django-celery-beat

在Django项目中，创建一个名为celery.py的文件，内容如下：

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'project_name.settings')  # 设置Django的环境变量

app = Celery('project_name')  # 创建Celery实例
app.config_from_object('django.conf:settings', namespace='CELERY')  # 读取Django的设置，以CELERY开头的

# 自动检索每个app下的tasks.py，添加到Celery中
app.autodiscover_tasks()

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

同时，我们还需要在settings.py中添加以下配置：

CELERY_BROKER_URL = 'amqp://localhost'  # RabbitMQ的Broker
CELERY_RESULT_BACKEND = 'django-db'  # Celery返回结果存放在Django中
CELERY_TIMEZONE = 'Asia/Shanghai'  # Celery时区设置

在settings.py还需要加入：

# 加入Celery Beat对时区的设置
CELERY_BEAT_SCHEDULE = {
    'task1': {
        'task': 'app_name.tasks.task1',
        'schedule': crontab(minute=0, hour='*/3')  # 设置任务的执行时间
    },
    'task2': {
        'task': 'app_name.tasks.task2',
        'schedule': crontab(minute='*/5')  #  * */5代表每5分钟执行一次
    },
}

CELERY_IMPORTS = (
    'app_name.tasks',
)

# 让django-celery-beat来控制Celery
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

# 注册应用tasks.py
CELERY_ALWAYS_EAGER = False
CELERY_IGNORE_RESULT = True

第四步：编写定时任务逻辑

我们需要在应用的tasks.py中自定义任务，例如：

from celery import shared_task

@shared_task
def task1():
    #  执行一些任务
    print('task1')

@shared_task
def task2():
    #  执行一些任务
    print('task2')

第五步：运行Celery Beat

在Django项目的根目录下，使用以下命令启动Celery Beat：

$ celery -A project_name worker -B --loglevel=info

最后，我们完成了Django与Celery Beat搭建定时任务的实现。

示例说明

下面是两个关于Celery Beat的示例：

示例一：定时发送邮件

假设我们有一个邮件发送的任务，我们可以使用Celery Beat来定时执行该任务。具体步骤如下（假设应用名称为myapp）：

1. 在myapp/tasks.py中定义任务：

from celery import shared_task
from django.core.mail import send_mail

@shared_task
def send_email():
    subject = 'Welcome to my site'
    message = 'Thank you for your registration'
    email_from = '[e-mail]'
    recipient_list = ['[e-mail]']
    send_mail(subject, message, email_from, recipient_list)

1. 在Django settings.py文件中，设置时间间隔

# 定时发送邮件任务
CELERY_BEAT_SCHEDULE = {
    'send-emails': {
        'task': 'myapp.tasks.send_email',
        'schedule': crontab(minute='*/30'),
    },
}

1. 最后，在命令行中运行Celery beat：

$ celery -A myproject beat -l info

示例二：定时生成缩略图

假设我们有一个图片处理的任务，我们可以使用Celery Beat来定时执行该任务。具体步骤如下（假设应用名称为myapp）：

1. 在myapp/tasks.py中定义任务

from PIL import Image
from celery import shared_task

@shared_task
def generate_thumbnail(image_path):
    with Image.open(image_path) as img:
        img.thumbnail((128, 128))
        img.save(f'thumbnail_{image_path}')

1. 在Django settings.py中，设置时间间隔

# 定时生成缩略图任务
CELERY_BEAT_SCHEDULE = {
    'generate-thumbnails': {
        'task': 'myapp.tasks.generate_thumbnail',
        'args': ('/path/to/image',),
        'schedule': crontab(hour='*/2'),
    },
}

1. 最后，在命令行中运行Celery beat：

$ celery -A myproject beat -l info

这两个例子展示了如何使用Celery Beat实现定时任务。你可以根据需要来定时执行不同的任务。