python django周期定时

在 Django 中，可以使用 Celery 和 Django-celery-beat 库来实现周期定时任务。

步骤如下：

1. 安装 Celery 和 Django-celery-beat 库：

pip install celery
pip install django-celery-beat

2. 在 Django 的 settings.py 文件中配置 Celery：

# settings.py

# Celery 配置
CELERY_BROKER_URL = 'amqp://guest@localhost//'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_BACKEND = 'django-db'
CELERY_TIMEZONE = 'Asia/Shanghai'

3. 在 Django 的 settings.py 文件中添加 Django-celery-beat 应用：

# settings.py

INSTALLED_APPS = [
    # ...
    'django_celery_beat',
]

4. 在 Django 的 urls.py 文件中添加 Celery 的 beat 路由：

# urls.py

from django.urls import path
from django_celery_beat.views import Beat


urlpatterns = [
    # ...
    path('celery-beat/', Beat.as_view(), name='celery-beat'),
]

5. 在 Django 的任意位置创建一个 tasks.py 文件，并定义周期任务：

# tasks.py

from celery import shared_task
from datetime import datetime


@shared_task
def test_task():
    print('当前时间：', datetime.now().strftime('%Y-%m-%d %H:%M:%S'))

6. 在 Django 的 settings.py 文件中配置周期任务：

# settings.py

CELERY_BEAT_SCHEDULE = {
    'test_task': {
        'task': 'tasks.test_task',
        'schedule': 30.0,  # 每 30 秒执行一次
    },
}

7. 启动 Celery：

celery -A your_project_name worker -l info
celery -A your_project_name beat -l info

8. 访问 http://127.0.0.1:8000/celery-beat/ 可以查看周期任务的执行记录。

注意：以上步骤中的 your_project_name 需要替换为你的项目名。