Django服务每日执行函数：实现令牌自动更新的最优方案

在Python/Django环境里，没法像JS那样用setInterval在Web进程里维持长期定时任务——因为Django的请求是短生命周期的，进程可能随时重启或回收。推荐用定时任务框架来实现每日更新令牌的需求，下面是几种成熟的实现方式：

方案一：Celery + Celery Beat（生产级首选） #

Celery是Python生态最成熟的异步任务框架，Celery Beat专门负责定时任务调度，适合高并发、多实例的生产环境。

1. 安装依赖 #

pip install celery redis

（Redis作为消息队列，也可替换为RabbitMQ）

2. 配置Celery #

在项目根目录创建celery.py：

import os
from celery import Celery
from celery.schedules import crontab

os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')

app = Celery('你的项目名')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# 配置每日定时任务（凌晨2点执行）
app.conf.beat_schedule = {
    'daily-refresh-expiring-tokens': {
        'task': '你的app名.tasks.refresh_expiring_tokens',
        'schedule': crontab(hour=2, minute=0),
    },
}

3. 定义刷新任务 #

在app目录下创建tasks.py：

from celery import shared_task
from django.utils import timezone
from .models import SomeModel
import requests

@shared_task
def refresh_expiring_tokens():
    # 筛选今日过期的令牌实例
    today = timezone.now().date()
    expiring_instances = SomeModel.objects.filter(
        refresh_token_expiry_date__date=today
    )
    
    for instance in expiring_instances:
        try:
            # 调用第三方API刷新令牌
            response = requests.post(
                '第三方API刷新地址',
                data={'refresh_token': instance.refresh_token}
            )
            response.raise_for_status()
            token_data = response.json()
            
            # 仅更新需要修改的字段，提升性能
            instance.access_token = token_data['access_token']
            instance.refresh_token = token_data['refresh_token']
            instance.refresh_token_expiry_date = timezone.now() + timezone.timedelta(weeks=2)
            instance.save(update_fields=['access_token', 'refresh_token', 'refresh_token_expiry_date'])
        except Exception as e:
            # 记录错误，避免单个实例失败中断整个任务
            print(f"实例{instance.id}令牌刷新失败: {str(e)}")

4. 启动服务 #

• 启动Redis服务
• 启动Celery Worker：celery -A 你的项目名 worker --loglevel=info
• 启动Celery Beat：celery -A 你的项目名 beat --loglevel=info

方案二：Django Q（轻量替代方案） #

如果不想引入Celery这类重型框架，Django Q是更轻量的选择，直接集成到Django生态中，无需额外的消息队列服务（可使用Django数据库作为任务存储）。

1. 安装配置 #

pip install django-q

在settings.py的INSTALLED_APPS中添加django_q，并配置任务集群：

Q_CLUSTER = {
    'name': 'DjangoORM',
    'workers': 4,
    'timeout': 90,
    'retry': 120,
    'queue_limit': 50,
    'orm': 'default'  # 使用Django数据库存储任务
}

2. 定义并注册定时任务 #

在app的tasks.py中：

from django_q.tasks import schedule
from django.utils import timezone
from .models import SomeModel
import requests

def refresh_expiring_tokens():
    today = timezone.now().date()
    expiring_instances = SomeModel.objects.filter(
        refresh_token_expiry_date__date=today
    )
    
    for instance in expiring_instances:
        try:
            response = requests.post(
                '第三方API刷新地址',
                data={'refresh_token': instance.refresh_token}
            )
            response.raise_for_status()
            token_data = response.json()
            
            instance.access_token = token_data['access_token']
            instance.refresh_token = token_data['refresh_token']
            instance.refresh_token_expiry_date = timezone.now() + timezone.timedelta(weeks=2)
            instance.save(update_fields=['access_token', 'refresh_token', 'refresh_token_expiry_date'])
        except Exception as e:
            print(f"实例{instance.id}令牌刷新失败: {str(e)}")

# 注册每日凌晨2点执行的任务
schedule(
    '你的app名.tasks.refresh_expiring_tokens',
    schedule_type='D',
    hour=2,
    minute=0
)

3. 启动任务集群 #

python manage.py qcluster

方案三：系统Cron + Django管理命令（无第三方依赖） #

如果不想引入任何异步框架，可直接用系统Cron定时执行Django管理命令，适合小型项目或资源受限的环境。

1. 创建管理命令 #

在app目录下创建management/commands/refresh_tokens.py（需先创建management和commands目录，每个目录都要有空的__init__.py）：

from django.core.management.base import BaseCommand
from django.utils import timezone
from ...models import SomeModel
import requests

class Command(BaseCommand):
    help = '每日刷新即将过期的令牌'

    def handle(self, *args, **options):
        today = timezone.now().date()
        expiring_instances = SomeModel.objects.filter(
            refresh_token_expiry_date__date=today
        )
        
        for instance in expiring_instances:
            try:
                response = requests.post(
                    '第三方API刷新地址',
                    data={'refresh_token': instance.refresh_token}
                )
                response.raise_for_status()
                token_data = response.json()
                
                instance.access_token = token_data['access_token']
                instance.refresh_token = token_data['refresh_token']
                instance.refresh_token_expiry_date = timezone.now() + timezone.timedelta(weeks=2)
                instance.save(update_fields=['access_token', 'refresh_token', 'refresh_token_expiry_date'])
                
                self.stdout.write(self.style.SUCCESS(f"实例{instance.id}令牌刷新成功"))
            except Exception as e:
                self.stderr.write(self.style.ERROR(f"实例{instance.id}令牌刷新失败: {str(e)}"))

2. 配置Cron #

运行crontab -e添加定时任务：

0 2 * * * /虚拟环境路径/bin/python /项目路径/manage.py refresh_tokens

（每天凌晨2点执行）

关键注意事项 #

• 批量优化：若实例数量较多，可分批次处理或使用批量更新操作，避免占用过多数据库连接。
• 锁机制：多实例部署时，需添加分布式锁（如Redis锁）或数据库锁，防止任务重复执行。
• 时区一致性：确保Django时区配置与第三方API时区一致，避免日期判断错误。
• 日志记录：建议使用Django的logging模块代替print，方便后续排查问题。

内容的提问来源于stack exchange，提问作者Mukhammaddiyor Alijonov