在Django中实现定时任务的多种方法

pip install celery
pip install django-celery-beat

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

# 设置默认的 Django 设置模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project_name.settings')

app = Celery('your_project_name')

# 从 Django 配置中读取 Celery 配置
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自动发现任务
app.autodiscover_tasks()

from __future__ import absolute_import, unicode_literals

# 确保任务模块被加载
from .celery import app as celery_app

__all__ = ('celery_app',)

# Celery 配置
CELERY_BROKER_URL = 'redis://localhost:6379/0'  # 使用 Redis 作为消息代理
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'  # 使用 Redis 作为结果存储
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'

from celery import shared_task

@shared_task
def my_periodic_task():
    # 执行定时任务的代码
    print("定时任务正在执行")

INSTALLED_APPS = [
    # 其他应用
    'django_celery_beat',
]

# 定时任务配置
CELERY_BEAT_SCHEDULE = {
    'my-task': {
        'task': 'my_app.tasks.my_periodic_task',
        'schedule': 3600.0,  # 每小时执行一次
    },
}

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

pip install django-background-tasks

INSTALLED_APPS = [
    # 其他应用
    'background_task',
]

from background_task import background

@background(schedule=60)
def my_periodic_task():
    # 执行定时任务的代码
    print("定时任务正在执行")

python manage.py process_tasks

from my_app.tasks import my_periodic_task

# 调度任务，每隔一分钟执行一次
my_periodic_task(repeat=60)

pip install apscheduler

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
import logging

logger = logging.getLogger(__name__)

def my_periodic_task():
    # 执行定时任务的代码
    print("定时任务正在执行")

scheduler = BackgroundScheduler()
scheduler.add_job(my_periodic_task, IntervalTrigger(seconds=3600))
scheduler.start()

# 确保在 Django 进程终止时关闭调度器
import atexit
atexit.register(lambda: scheduler.shutdown())

from django.apps import AppConfig

class MyAppConfig(AppConfig):
    name = 'my_app'

    def ready(self):
        import my_app.scheduler