如何在Django中实现用户专属的定时递增后台任务？

1. 环境配置 #

• 安装依赖包：pip install django-background-task
• 在settings.py的INSTALLED_APPS中添加：'background_task',
• 执行数据库迁移：python manage.py migrate

2. 创建数据模型 #

在你的app的models.py中定义存储用户数值与任务状态的模型：

from django.db import models
from django.contrib.auth.models import User

class UserNumber(models.Model):
    user = models.OneToOneField(User, on_delete=models.CASCADE, related_name='user_number')
    current_value = models.FloatField(default=0.0)
    is_running = models.BooleanField(default=False)
    start_time = models.DateTimeField(null=True, blank=True)

    def __str__(self):
        return f"{self.user.username}: {self.current_value}"

3. 编写后台任务 #

在app目录下新建tasks.py，实现递归调度的更新任务：

from background_task import background
from django.utils import timezone
from .models import UserNumber

@background(schedule=1)
def update_user_number(user_id):
    try:
        user_num = UserNumber.objects.get(user_id=user_id)
        # 检查任务是否已运行满1天（86400秒）
        if (timezone.now() - user_num.start_time).total_seconds() >= 86400:
            user_num.is_running = False
            user_num.save()
            return
        
        # 每秒增加当前数值的1%（复利计算）
        user_num.current_value *= 1.01
        user_num.save()
        
        # 1秒后调度下一次更新
        update_user_number(user_id, schedule=1)
    except UserNumber.DoesNotExist:
        pass

4. 实现用户提交与状态视图 #

在views.py中处理用户输入提交和数值状态展示：

from django.shortcuts import render, redirect
from django.contrib.auth.decorators import login_required
from .forms import NumberInputForm
from .models import UserNumber
from .tasks import update_user_number
from django.utils import timezone

@login_required
def submit_number(request):
    if request.method == 'POST':
        form = NumberInputForm(request.POST)
        if form.is_valid():
            initial_value = form.cleaned_data['initial_value']
            user_num, created = UserNumber.objects.get_or_create(user=request.user)
            if not user_num.is_running:
                user_num.current_value = initial_value
                user_num.is_running = True
                user_num.start_time = timezone.now()
                user_num.save()
                # 启动第一个更新任务
                update_user_number(request.user.id)
            return redirect('number_status')
    else:
        form = NumberInputForm()
    return render(request, 'submit_number.html', {'form': form})

@login_required
def number_status(request):
    user_num = UserNumber.objects.get(user=request.user)
    return render(request, 'number_status.html', {'user_num': user_num})

5. 定义输入表单 #

在forms.py中实现数值输入验证：

from django import forms

class NumberInputForm(forms.Form):
    initial_value = forms.FloatField(label='初始数值', min_value=0.01)

6. 启动任务处理器 #

打开终端执行命令，启动后台任务worker：

python manage.py process_tasks

如果坚持使用Celery，大概率是此前配置步骤缺失，按以下流程重新配置：

1. 环境配置 #

• 安装依赖：pip install celery redis（用Redis作为消息代理，上手门槛更低）
• 在项目根目录新建celery.py：

import os
from celery import Celery

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

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

• 在项目根目录的__init__.py中添加：

from .celery import app as celery_app

__all__ = ('celery_app',)

• 在settings.py中添加Celery配置：

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_TIMEZONE = 'Asia/Shanghai'

2. 编写Celery任务 #

在tasks.py中实现循环更新的任务逻辑：

from celery import shared_task
from django.utils import timezone
from .models import UserNumber
import time

@shared_task
def update_user_number_task(user_id):
    user_num = UserNumber.objects.get(user_id=user_id)
    end_time = user_num.start_time + timezone.timedelta(days=1)
    
    while timezone.now() < end_time and user_num.is_running:
        user_num.current_value *= 1.01
        user_num.save()
        time.sleep(1)
        # 重新获取实例，避免缓存导致的数值不更新
        user_num = UserNumber.objects.get(user_id=user_id)
    
    user_num.is_running = False
    user_num.save()

3. 视图中启动Celery任务 #

修改submit_number视图中的任务启动代码：

# 导入Celery任务
from .tasks import update_user_number_task

# ... 其他代码
if not user_num.is_running:
    user_num.current_value = initial_value
    user_num.is_running = True
    user_num.start_time = timezone.now()
    user_num.save()
    # 异步启动任务
    update_user_number_task.delay(request.user.id)
# ...

4. 启动Celery Worker #

打开终端执行命令：

celery -A 你的项目名 worker --loglevel=info

• django-background-task的任务存储在数据库中，重启worker后未完成的任务会自动续跑，适合小体量场景
• Celery方案中若重启worker，正在运行的任务会中断；如需高可靠性，可搭配Celery Beat做定时调度，但每秒调度会增加数据库压力
• 每秒更新一次数据库存在性能损耗，若前端仅需展示实时数值，可考虑后端存储初始值与开始时间，前端用JS计算显示，后端仅在任务结束时更新最终值——但按需求要后台实时更新数值，仍需写入数据库
• 必须添加登录验证，确保每个用户的任务与数值独立隔离

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