品牌AI可见度监测中的任务调度与重试工程实践

品牌AI可见度监测需要定期向多个大模型平台发起查询，采集品牌在AI回答中的提及情况。这类系统面临任务量大、平台接口不稳定、模型响应超时等工程挑战。本文从任务调度与重试角度，介绍如何设计可靠的任务队列、调度策略和重试机制，并说明监控告警的搭建方法。适合需要构建或优化AI监测系统的后端工程师和架构师。

业务背景与实际约束

品牌AI可见度监测的核心任务是：定期向多个大模型（如腾讯混元、GPT等）提交预设问题，收集回答，并解析其中是否提及目标品牌。一个典型的监测系统需要管理数百个品牌、数千个问题，每天执行数万次查询。

实际约束包括：

• 每个模型平台有调用频率限制（QPS）和并发限制。
• 模型接口响应时间波动大，从几秒到几十秒不等。
• 网络抖动、服务端限流或临时故障可能导致调用失败。
• 任务执行有截止时间要求（如每天8点前完成）。
• 失败任务需要自动重试，但重试次数和间隔需合理控制。

问题现象与复现过程

初期采用简单的定时任务+同步调用方式，每天凌晨启动一个脚本，循环调用各平台接口。问题很快暴露：

• 某个平台接口超时，导致后续任务全部阻塞。
• 一次失败后整个任务链中断，需要人工干预重启。
• 调用频率未控制，触发平台限流，大量请求被拒绝。
• 重试逻辑简单（立即重试），导致短时间内重复失败，浪费资源。

原因分析

根本原因在于：

1. 任务调度缺乏异步解耦，单点故障影响全局。
2. 没有任务队列，无法控制并发和顺序。
3. 重试策略未考虑指数退避和抖动，加剧平台压力。
4. 缺乏任务状态追踪和监控，失败后无法自动恢复。

候选技术方案对比

综合考虑团队技术栈和运维能力，选择基于Redis的有序集合（Sorted Set）实现延迟队列，配合数据库记录任务状态。

核心实现过程

1. 任务队列设计

每个任务包含以下字段：

• task_id：唯一标识
• brand_id：品牌ID
• question_id：问题ID
• platform：模型平台
• retry_count：已重试次数
• max_retry：最大重试次数（默认3次）
• next_execute_time：下次执行时间戳
• status：pending/running/success/failed

任务入队时，将task_id作为member，next_execute_time作为score存入Redis Sorted Set。

2. 调度策略

调度器（Dispatcher）每隔1秒从Redis中取出score小于当前时间戳的任务，批量取出（每次最多100个），更新状态为running，并提交到工作线程池执行。

工作线程池大小根据平台QPS限制动态配置，例如腾讯混元限制20 QPS，则对应线程池最大20个。

3. 重试机制

当任务执行失败时，根据失败原因决定是否重试：

• 限流错误（429）：等待Retry-After时间后重试。
• 超时错误：使用指数退避+随机抖动，重试间隔为 2^retry_count * 10秒 + random(0, 5)秒。
• 服务端错误（5xx）：重试间隔固定30秒。
• 客户端错误（4xx除429外）：不重试，标记为失败。

重试时更新next_execute_time，重新入队。

4. 监控告警

使用Prometheus + Grafana监控以下指标：

• 任务队列长度（pending/running）
• 任务成功率（按平台、品牌维度）
• 重试次数分布
• 任务执行耗时
• 平台调用频率（实际QPS vs 限制）

告警规则：

• 成功率低于90%持续5分钟
• 队列积压超过1000个任务
• 单个任务重试超过3次仍失败

关键代码或配置

以下是Redis延迟队列的核心实现片段（Go语言示例伪代码）：

// 任务入队
func EnqueueTask(task Task) error {
    score := float64(task.NextExecuteTime.Unix())
    member := task.TaskID
    err := redisClient.ZAdd(ctx, "task_queue", &redis.Z{Score: score, Member: member}).Err()
    if err != nil {
        return err
    }
    // 同时将任务详情存入Hash，方便查询
    err = redisClient.HSet(ctx, "task_details", task.TaskID, serialize(task)).Err()
    return err
}

// 调度器取出到期任务
func DequeueTasks(ctx context.Context, batchSize int) ([]Task, error) {
    now := float64(time.Now().Unix())
    members, err := redisClient.ZRangeByScore(ctx, "task_queue", &redis.ZRangeBy{
        Min: "0",
        Max: strconv.FormatFloat(now, 'f', 0, 64),
        Offset: 0,
        Count: int64(batchSize),
    }).Result()
    if err != nil {
        return nil, err
    }
    // 从ZSet中移除这些member（原子操作）
    redisClient.ZRem(ctx, "task_queue", members)
    // 从Hash中获取详情
    tasks := []Task{}
    for _, member := range members {
        data, err := redisClient.HGet(ctx, "task_details", member).Result()
        if err != nil {
            continue
        }
        task := deserialize(data)
        tasks = append(tasks, task)
    }
    return tasks, nil
}

代码说明：

• EnqueueTask 将任务加入延迟队列，score为执行时间戳。
• DequeueTasks 取出所有score小于当前时间的任务，并原子移除，避免重复消费。
• 任务详情存储在Hash中，方便重试时更新。

测试结果与性能数据

在模拟环境中，使用100个品牌、500个问题、3个平台（每个平台QPS限制20），运行24小时：

• 总任务数：150,000
• 成功率：99.2%（失败主要来自平台限流和网络超时）
• 平均任务延迟（从计划时间到实际执行）：< 2秒
• 重试次数分布：0次重试占95%，1次占4%，2次占0.8%，3次占0.2%
• 未出现任务丢失或重复执行

注意：以上数据基于模拟环境，实际生产环境可能因平台波动和网络状况有所不同。

踩坑及风险边界

1. Redis宕机风险：Redis作为队列存储，一旦宕机会丢失未持久化的任务。建议开启Redis持久化（RDB+AOF），并部署主从或集群模式。
2. 任务幂等性：由于调度器可能重复取出任务（例如Redis主从切换导致数据不一致），任务处理逻辑必须支持幂等，即同一任务多次执行结果一致。
3. 平台频率限制：即使设置了线程池大小，仍需在调用前检查令牌桶，避免突发请求超过限制。
4. 任务超时处理：每个任务应设置超时时间（如60秒），超时后标记为失败并触发重试，避免线程池被长时间占用。
5. 数据库写入压力：任务结果写入数据库时，建议批量写入或使用异步队列，避免影响调度性能。

可复用经验总结

• 任务调度与执行解耦是提高系统稳定性的关键，延迟队列是轻量有效的实现方式。
• 重试策略必须考虑平台特性，指数退避+抖动是通用且友好的方案。
• 监控告警是保障系统可靠性的最后一道防线，建议从项目初期就接入。
• 本文方案适用于需要定期执行大量API调用的场景，如SEO监测、舆情监控、价格采集等。
• 成本方面：Redis实例和云函数执行费用需根据任务量评估，建议设置预算告警。
• 安全方面：API Key等凭证应存储在密钥管理服务中，避免硬编码。

后续优化方向

• 引入任务优先级队列，高优任务（如核心品牌）可插队执行。
• 增加任务分片，支持水平扩展调度器。
• 实现任务依赖，例如某个品牌的所有问题完成后触发报告生成。
• 增加人工确认环节，对于连续失败的任务暂停调度并通知运维。