django_celery_beat.schedulers.DatabaseScheduler
时间: 2025-06-29 19:15:09 浏览: 14
### Django Celery Beat DatabaseScheduler 配置与使用
#### 安装依赖库
为了使 `Django` 和 `Celery Beat` 的集成更加顺畅,需安装必要的 Python 库。这通常涉及安装 `celery`, `django-celery-beat` 及其相关依赖项。
```bash
pip install django-celery-beat celery
```
#### 设置 Celery 实例
创建一个名为 `celery.py` 文件,在项目根目录下的应用包内定义 Celery 实例并设置 Broker URL:
```python
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
app = Celery('your_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
```
#### 启用数据库调度器
在项目的 settings 中指定 `beat_scheduler` 参数来启用 `DatabaseScheduler`. 这样可以利用 Django ORM 来管理周期性任务.
```python
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers.DatabaseScheduler'
```
#### 创建周期性任务
通过 Django Admin 或者编程方式向 `PeriodicTask` 表中添加新条目以安排定期执行的任务:
```python
from datetime import timedelta
from django_celery_beat.models import PeriodicTask, IntervalSchedule
schedule, created = IntervalSchedule.objects.get_or_create(
every=10,
period=IntervalSchedule.SECONDS,
)
task = PeriodicTask.objects.create(
interval=schedule, # 使用上面创建的时间间隔对象
name='Test Task', # 名字唯一
task='my_app.tasks.test_task', # 路径到任务函数
)
```
上述代码片段展示了如何每十秒钟运行一次位于 `my_app/tasks.py` 下名称为 `test_task()` 函数的方法[^1].
#### 执行命令启动服务
最后一步是在终端里分别启动 Celery worker 和 beat scheduler 两个进程:
```bash
# Terminal Window 1 (Worker Process)
celery -A your_project worker --loglevel=info
# Terminal Window 2 (Beat Scheduler Process)
celery -A your_project beat --loglevel=info
```
阅读全文
相关推荐


















