pip install celery
pip install django-celery-beat

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

# 設(shè)置默認(rèn)的 Django 設(shè)置模塊
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')

# 自動(dòng)發(fā)現(xiàn)任務(wù)
app.autodiscover_tasks()

from __future__ import absolute_import, unicode_literals

# 確保任務(wù)模塊被加載
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 作為結(jié)果存儲(chǔ)
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():
    # 執(zhí)行定時(shí)任務(wù)的代碼
    print("定時(shí)任務(wù)正在執(zhí)行")

INSTALLED_APPS = [
    # 其他應(yīng)用
    'django_celery_beat',
]

# 定時(shí)任務(wù)配置
CELERY_BEAT_SCHEDULE = {
    'my-task': {
        'task': 'my_app.tasks.my_periodic_task',
        'schedule': 3600.0,  # 每小時(shí)執(zhí)行一次
    },
}

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

pip install django-background-tasks

INSTALLED_APPS = [
    # 其他應(yīng)用
    'background_task',
]

from background_task import background

@background(schedule=60)
def my_periodic_task():
    # 執(zhí)行定時(shí)任務(wù)的代碼
    print("定時(shí)任務(wù)正在執(zhí)行")

python manage.py process_tasks

from my_app.tasks import my_periodic_task

# 調(diào)度任務(wù)，每隔一分鐘執(zhí)行一次
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():
    # 執(zhí)行定時(shí)任務(wù)的代碼
    print("定時(shí)任務(wù)正在執(zhí)行")

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

# 確保在 Django 進(jìn)程終止時(shí)關(guān)閉調(diào)度器
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