如何在Django应用中实现启动后定时更新SingleProduct表的无限循环任务

一、数据库模型设计 #

首先定义SingleProduct表，确保仅存储一条当前展示的产品记录：

# your_app/models.py
from django.db import models

class Product(models.Model):
    name = models.CharField(max_length=255)
    price = models.DecimalField(max_digits=10, decimal_places=2)
    rating = models.FloatField()
    # 其他业务字段...

class SingleProduct(models.Model):
    product = models.OneToOneField(Product, on_delete=models.CASCADE)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        verbose_name = "当前展示产品"
        verbose_name_plural = "当前展示产品"

二、定时任务实现 #

推荐使用APScheduler实现后台定时任务（轻量适合小项目），替代手动写无限循环（避免阻塞Django进程）。

1. 安装依赖 #

pip install apscheduler

2. 编写任务函数 #

创建tasks.py文件，实现随机产品更新逻辑：

# your_app/tasks.py
import random
import logging
from django.db import transaction
from django.core.cache import cache
from .models import Product, SingleProduct

logger = logging.getLogger(__name__)

def update_featured_product():
    try:
        # 优化随机产品查询（替代order_by('?')，避免大数据量表性能问题）
        max_product_id = Product.objects.aggregate(models.Max('id'))['id__max']
        if not max_product_id:
            return
        
        random_product = None
        while not random_product:
            random_id = random.randint(1, max_product_id)
            random_product = Product.objects.filter(id=random_id).first()
        
        # 原子更新SingleProduct，确保仅一条记录
        with transaction.atomic():
            featured, created = SingleProduct.objects.get_or_create(
                defaults={'product': random_product}
            )
            if not created:
                featured.product = random_product
                featured.save()
        
        # 更新后清除缓存（若启用缓存）
        cache.delete('featured_product')
    except Exception as e:
        logger.error(f"更新展示产品失败: {str(e)}")

3. 启动调度器 #

在应用的apps.py中配置启动时执行任务并开启定时调度：

# your_app/apps.py
import os
from django.apps import AppConfig
from apscheduler.schedulers.background import BackgroundScheduler
from .tasks import update_featured_product

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app'

    def ready(self):
        # 仅在主进程启动时执行（避免开发环境runserver多进程重复启动）
        if os.environ.get('RUN_MAIN') == 'true':
            # 应用启动立即执行一次
            update_featured_product()
            # 每15分钟执行一次更新
            scheduler = BackgroundScheduler()
            scheduler.add_job(update_featured_product, 'interval', minutes=15)
            scheduler.start()

记得在settings.py中替换默认的应用配置：

# settings.py
INSTALLED_APPS = [
    # ... 其他应用 ...
    'your_app.apps.YourAppConfig',
]

三、首页展示实现 #

在视图中获取当前展示产品，可配合缓存减少数据库查询：

# your_app/views.py
from django.shortcuts import render
from django.core.cache import cache
from .models import SingleProduct

def home(request):
    # 从缓存获取，缓存15分钟与任务间隔同步
    featured_product = cache.get('featured_product')
    if not featured_product:
        single_product = SingleProduct.objects.first()
        featured_product = single_product.product if single_product else None
        cache.set('featured_product', featured_product, 60*15)
    
    return render(request, 'home.html', {'featured_product': featured_product})

模板中渲染产品：

<!-- templates/home.html -->
{% if featured_product %}
<div class="featured-product-card">
    <h3>今日随机推荐</h3>
    <p>产品名称: {{ featured_product.name }}</p>
    <p>价格: ¥{{ featured_product.price }}</p>
    <p>评分: {{ featured_product.rating }}/5</p>
</div>
{% else %}
<p>暂无可展示产品</p>
{% endif %}

四、核心注意要点 #

• 原子操作保障：用transaction.atomic()包裹SingleProduct的更新逻辑，避免并发场景下出现多条记录或更新失败。
• 避免重复调度：开发环境下runserver会启动两个进程，需通过RUN_MAIN环境变量判断仅主进程启动调度器；生产环境部署时（如Gunicorn），确保仅一个进程启动调度，或用文件锁防止重复。
• 查询性能优化：大数据量的Product表中，避免使用order_by('?')，改用随机ID查询的方式减少数据库压力。
• 异常处理与日志：任务函数中添加异常捕获和日志记录，避免单个任务失败导致调度器崩溃。
• 缓存配合：首页展示时用Django缓存存储产品数据，减少数据库查询次数，提升页面响应速度。
• 生产环境进程管理：若使用后台调度器，生产环境需确保调度进程稳定运行，可配合进程管理工具（如Supervisor）监控。

内容的提问来源于stack exchange，提问作者some nooby questions