使用celery-beat作为定时器

Question

在我的应用程序中，我想在创建对象后的一定时间间隔后运行某个函数。 django-celery-beat 是执行此操作的正确工具吗？

Answer 1

如果您正在考虑使用 django-celery-beat 但只有几个任务需要安排，那么设置 Celery 可能会让人感觉有点大材小用，因为配置有时会非常耗时。在这种情况下，django-rq 提供了一种更简单的替代方案，使用 Redis 作为消息代理为后台任务提供类似的功能。

以下是如何开始使用 django-rq：

1. 安装 django-rq

pip install django-rq

2. 将 django_rq 添加到 INSTALLED_APPS

已安装的应用程序 = [ # 其他应用程序... 'django_rq'，]

3. 为 django-rq 配置 Redis 设置 将此配置添加到您的设置文件中：

RQ_QUEUES = {
    'default': {
        'HOST': 'localhost',
        'PORT': 6379,
        'DB': 0,
        'DEFAULT_TIMEOUT': 360,
    }
}

4. 定义任务 定义一个函数来执行您想要安排的任务。例如，以下任务通过 ID 获取对象并执行指定的操作：

# tasks.py in your app
from myapp.models import MyModel

def my_scheduled_task(object_id):
    obj = MyModel.objects.get(id=object_id)
    # Perform the desired action here
    print(f"Running task for object {obj}")

5. 保存新模型实例时安排任务 下面是一个使用 Django 模型 save() 方法在创建新对象 24 小时后安排任务的示例：

# models.py in your app
from django.db import models
from django.utils import timezone
from datetime import timedelta
import django_rq
from myapp.tasks import my_scheduled_task

class MyModel(models.Model):
    created_at = models.DateTimeField(auto_now_add=True)

    def save(self, *args, **kwargs):
        is_new = self.pk is None
        super().save(*args, **kwargs)

        # Schedule the task only if this is a new object
        if is_new:
            scheduler = django_rq.get_scheduler('default')
            run_at = timezone.now() + timedelta(hours=24)
            scheduler.enqueue_at(run_at, my_scheduled_task, self.id)

6. 运行 Worker 和 Scheduler 在单独的终端窗口中，运行以下命令：

python manage.py rqworker 默认值

python 管理.py rqscheduler

总结 django-rq 提供了一个轻量级、Redis 支持的任务队列，非常适合小型任务，无需 Celery 的完整配置。它以最少的设置有效地处理计划任务和后台任务。