Celery调度最佳实践：千级企业数据更新任务规划咨询

问题描述 #

我维护的系统里有Company实体，每个企业对应某服务的API_token，用来拉取数据存入数据库，用户可以对这些token进行增删改操作。现在企业数量已经超过1000家，我需要设计合理的任务调度方案来执行数据更新，目前想到两种思路：

1. 单调度任务：创建一个定时任务，遍历所有企业执行数据更新
2. 独立调度任务：为每个企业单独创建定时任务

想请教这两种方案的优劣，或者有没有更优的实现方式。当前的代码示例如下：

@shared_task()
def update_wb_statistic(company: Company) -> None:
    if company.wbToken:
        company_api = WildberriesAPI(company)
        company_api.get_incomes()
        company_api.get_sales()
        company_api.get_stocks()
        company.wbLastUpdate = datetime.now()


@shared_task()
def update_companies():
    companies = Company.query.all()
    for company in companies:
        if company.wbToken:
            update_wb_statistic.delay(company)

两种方案的优劣分析 #

方案1：单调度任务（批量触发子任务） #

优势

• 管理成本低：只需要维护一个定时任务，用户增删改token时不用同步调整调度，省掉大量重复操作
• 资源易管控：通过任务队列的并发配置（比如Celery的worker数量）就能统一控制整体更新的并发量，避免瞬间冲爆目标API
• 逻辑集中：批量任务里可以统一过滤无token的企业，不用在每个调度任务里重复写判断逻辑

劣势

• 单点风险高：如果这个批量任务本身执行失败（比如查询企业列表时数据库出错），所有企业的更新都会停摆
• 内存压力大：企业数涨到几万级时，Company.query.all()会一次性把所有数据拉到内存，容易引发内存溢出
• 请求过于集中：所有子任务几乎同时触发，短时间内对目标API发起大量请求，很容易触发对方的限流甚至封禁

方案2：为每个企业单独创建调度任务 #

优势

• 故障隔离性强：单个企业的调度任务失败，不会影响其他企业的更新
• 时间配置灵活：可以给不同企业设置不同的更新频率（比如有的企业需要每小时更，有的每天更），适配个性化需求
• 内存压力分散：每个任务只处理单个企业，不会出现批量加载数据的内存占用问题

劣势

• 维护成本爆炸：1000+企业就要维护1000+定时任务，用户增删改token时，还要同步创建/删除/修改对应的调度，逻辑复杂度直接拉满
• 资源难统一管控：大量任务同时触发的话，不仅可能冲爆目标API，自身系统的CPU、内存也会被占满
• 调度系统负载高：过多的定时任务会给Celery Beat、APScheduler这类调度组件带来额外负载，容易出现调度延迟或任务遗漏的情况

更优方案：分片批量+动态调度的混合模式 #

结合两种方案的优点，推荐以下实现思路：

1. 分片批量触发：把企业按ID分片（比如每50个一组），创建多个批量任务，每个任务负责一组企业的子任务触发，既避免单批量任务的单点风险，又降低内存压力
2. 分散请求峰值：给每个子任务加0-30秒的随机延迟，避免所有请求同时打向目标API
3. 事件触发补充更新：用户修改企业token时，立即触发一次该企业的更新，保证数据及时性
4. 失败自动重试：给单个企业的更新任务加重试机制，处理API临时不可用的情况

优化后的代码示例：

from celery import shared_task
from django.db.models import Q
import random
from time import sleep
from datetime import datetime

@shared_task(autoretry_for=(Exception,), retry_backoff=3, retry_kwargs={"max_retries": 3})
def update_wb_statistic(company_id: int) -> None:
    # 改用ID查询，避免传递整个对象的序列化问题
    company = Company.objects.get(id=company_id)
    if company.wbToken:
        company_api = WildberriesAPI(company)
        company_api.get_incomes()
        company_api.get_sales()
        company_api.get_stocks()
        company.wbLastUpdate = datetime.now()
        company.save()  # 别忘了保存修改

@shared_task
def update_companies_batch(start_id: int, end_id: int):
    # 按ID范围分片查询，直接过滤有token的企业
    companies = Company.objects.filter(
        Q(id__gte=start_id) & 
        Q(id__lte=end_id) & 
        Q(wbToken__isnull=False)
    )
    for company in companies:
        # 加随机延迟，分散请求
        sleep(random.randint(0, 30))
        update_wb_statistic.delay(company.id)

@shared_task
def dispatch_update_tasks():
    # 获取企业ID的范围，计算分片
    id_range = Company.objects.aggregate(min_id=models.Min('id'), max_id=models.Max('id'))
    min_id, max_id = id_range['min_id'], id_range['max_id']
    if not min_id or not max_id:
        return
    
    batch_size = 50  # 每批处理50个企业，可根据实际调整
    current_start = min_id
    while current_start <= max_id:
        current_end = min(current_start + batch_size - 1, max_id)
        update_companies_batch.delay(current_start, current_end)
        current_start = current_end + 1

额外提醒 #

• 别直接传Company对象到任务里，容易引发序列化问题，用ID查询更稳妥
• 加个任务监控和告警，比如用Celery Flower，或者自定义失败告警，及时处理异常企业的更新
• 在WildberriesAPI里加请求频率控制，比如用重试库做限流，避免被目标服务封禁

内容的提问来源于stack exchange，提问作者Сервер Чауш