本文介绍了如何在 Django 项目中实现定时删除数据库记录的功能，重点讲解了使用 Celery 作为后台任务队列，定期检查并删除满足特定条件的旧数据，以保持数据库的整洁和性能。通过本文，你将学习如何配置 Celery，创建定时任务，以及编写删除过期数据的代码。
在 Django 项目中，定期清理过期数据是一项常见的需求，例如删除过期的用户会话、日志记录或临时文件。虽然 Django Signals 可以用于在模型实例被删除前后执行操作，但它们并不适合处理定时任务。更可靠和高效的方法是使用 Celery 这样的后台任务队列，它可以定期执行任务，而不会阻塞主应用程序的运行。
1. Celery 简介与安装
Celery 是一个强大的分布式任务队列，可以异步地执行任务。它常用于处理耗时的操作，例如发送邮件、处理图像或定期清理数据。
首先，需要安装 Celery 及其依赖项：
pip install celery redis
这里我们使用 Redis 作为 Celery 的消息代理（broker），你也可以选择其他的消息代理，例如 RabbitMQ。
2. 配置 Celery
在你的 Django 项目中，创建一个 celery.py 文件，通常放在与 settings.py 同级的目录下：
# celery.py
import os
from celery import Celery
# 设置 Django 的 settings 模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project_name.settings') # 将 your_project_name 替换为你的项目名
app = Celery('your_project_name') # 将 your_project_name 替换为你的项目名
# 从 Django 的 settings 文件中加载 Celery 配置
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自动发现 tasks.py 文件
app.autodiscover_tasks()
@app.task(bind=True)
def debug_task(self):
print(f'Request: {self.request!r}')接下来，在你的 settings.py 文件中添加 Celery 的配置：
# settings.py CELERY_BROKER_URL = 'redis://localhost:6379/0' # Redis 连接 URL CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' # 存储任务结果的 backend CELERY_ACCEPT_CONTENT = ['application/json'] CELERY_TASK_SERIALIZER = 'json' CELERY_RESULT_SERIALIZER = 'json' CELERY_TIMEZONE = 'Asia/Shanghai' # 设置时区，根据你的实际情况修改
最后，在你的 __init__.py 文件中导入 Celery 应用：
# __init__.py
from .celery import app as celery_app
__all__ = ('celery_app',)3. 创建定时任务
在你的 Django 应用中（例如 smart_search），创建一个 tasks.py 文件：
# smart_search/tasks.py
from celery import shared_task
from django.utils import timezone
from datetime import timedelta
from .models import UserHitCount
@shared_task
def delete_old_user_hit_counts():
"""
删除创建时间超过 15 天的 UserHitCount 记录。
"""
cutoff_date = timezone.now() - timedelta(days=15)
UserHitCount.objects.filter(created_at__lte=cutoff_date).delete()
print(f"Deleted UserHitCount records created before {cutoff_date}")这个任务使用 shared_task 装饰器将其注册为 Celery 任务。它查询 UserHitCount 模型，找到所有 created_at 早于 15 天前的记录，并将其删除。
4. 配置 Celery Beat (定时任务调度器)
Celery Beat 是一个调度器，用于定期执行 Celery 任务。在 settings.py 中配置 Celery Beat：
# settings.py
from celery.schedules import crontab
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler' # 使用数据库存储调度信息
CELERY_BEAT_SCHEDULE = {
'delete_old_user_hit_counts': {
'task': 'smart_search.tasks.delete_old_user_hit_counts',
'schedule': crontab(hour=0, minute=0), # 每天凌晨 0 点执行
},
}这里，我们使用 django-celery-beat 插件将调度信息存储在数据库中。你需要安装这个插件：
pip install django-celery-beat
然后，将 django_celery_beat 添加到你的 INSTALLED_APPS 中：
# settings.py
INSTALLED_APPS = [
# ...
'django_celery_beat',
# ...
]并运行迁移：
python manage.py migrate
5. 启动 Celery 和 Celery Beat
打开三个终端窗口，分别启动 Redis, Celery Worker 和 Celery Beat：
-
启动 Redis:
redis-server
-
启动 Celery Worker:
celery -A your_project_name worker -l info # 将 your_project_name 替换为你的项目名
-
启动 Celery Beat:
celery -A your_project_name beat -l info -s celerybeat-schedule # 将 your_project_name 替换为你的项目名
6. 测试定时任务
现在，Celery Beat 将按照你配置的计划定期执行 delete_old_user_hit_counts 任务。你可以在 Celery Worker 的日志中看到任务的执行情况。
你也可以手动触发任务进行测试：
from smart_search.tasks import delete_old_user_hit_counts delete_old_user_hit_counts.delay()
这会将任务放入 Celery 队列中，Celery Worker 会立即执行它。
注意事项和总结
- 错误处理: 在 Celery 任务中添加适当的错误处理机制，例如使用 try...except 块来捕获异常，并记录错误信息。
- 并发: Celery 允许你配置并发 Worker 的数量。根据你的服务器资源和任务负载，调整并发 Worker 的数量以获得最佳性能。
- 数据库连接: 确保 Celery Worker 可以访问你的 Django 数据库。
- 时区: 确保 Celery 和 Django 的时区设置一致，以避免定时任务执行时间不准确的问题。
通过使用 Celery，你可以轻松地实现 Django 项目中的定时任务，例如定期清理数据库记录。这种方法比使用 Django Signals 更加可靠和高效，并且可以避免阻塞主应用程序的运行。记得根据你的实际需求配置 Celery 和 Celery Beat，并添加适当的错误处理机制，以确保定时任务的稳定运行。
