从简陋队列到可恢复的任务队列
这篇文章讲的不是如何实现一个更复杂的队列，而是当任务会花钱、会中断、会重启时，队列周围需要补上哪些状态和规则。
简陋队列
- 把任务放进队列，再由消费者逐个执行，是理解任务队列最直接的方式：
$queue[] = $job;
while (!empty($queue)) {
$job = array_shift($queue);
execute($job);
}这个简单队列的缺点
- 没有保存任务：队列只在内存中，程序退出后就会丢失。
- 没有记录进度：不知道哪些任务完成了、哪些失败了。
- 没有异常处理：执行任务时抛出未捕获异常，会中断后面的任务。
- 没有恢复机制：失败或重启后，无法根据记录继续处理。
- 只能逐个执行：前面的任务耗时较长，后面的任务就只能等待。
优化点概览
- 任务队列的优化，不只是让任务执行得更快，还要让执行过程有记录、失败有处理、中断后有恢复依据。
1.任务状态与持久化：记录每条帖子的处理状态，并保存到本地。
2.批次与并发控制：帖子分批处理，限制同时运行的批次数量。
3.分类失败与有限补试：根据失败原因选择等待、补试或结束。
4.熔断与保存失败保护：发生连续传输故障或核心状态保存失败时，限制后续执行。
5.中断恢复与结果复用：根据已有输出和上游任务 ID，选择合适的恢复方式。
6.进度与风险反馈：展示真实进度、失败原因，并在必要时要求用户确认- 主要状态
pending 等待安排
queued 已经排队
waiting 等待指定时间后再尝试
running 正在执行
received 已保存输出，等待校验
completed 校验通过并保存
failed 处理失败1. 任务状态与持久化
如果只使用内存队列，程序退出后，就很难判断任务处理到了哪里。
因此，项目按月份保存每条帖子的状态、尝试次数、结果和失败原因。队列负责安排执行，持久化记录负责保留进度。 一个批次执行前，先把对应帖子标记为 running，保存成功后才调用 AI。
下面用 PHP 表达这个先后关系
其中 $result 是月度记录 $snapshot 中的帖子结果对象：
$result->status = 'running';
$result->attempts += 1;
// Persist before calling AI.
persist($snapshot);
$output = callModel($posts);
收到输出后，先保存为 received；结果校验通过并保存后，才变成 completed。
这样，即使程序在校验过程中退出，也有机会利用已经保存的输出继续处理。
同一个月可能有多个批次同时返回，所以这个月份的修改和保存需要串行执行，避免相互覆盖。
需要注意，running 只代表本地开始执行，不代表上游一定收到了请求。2. 批次与并发控制
- 帖子较多时，不适合一次性全部交给 AI。项目先将帖子拆成批次，当前应用配置为每批最多 25 条，同时还限制文本长度。
- 分批之后，也不同时启动所有批次，而是正常情况下最多运行两个批次。一个批次结束后释放名额，再启动后面的批次。
- 可以把它理解为：
有两个执行名额，拿到名额的批次才能开始，其他批次继续等待。这个名额覆盖创建任务、等待结果、校验和保存的整个过程，
不是 HTTP 请求发出去就立即释放。
无论成功还是失败，批次结束后都要释放名额。另外，先启动的批次不一定先完成。队列控制的是启动顺序，不是上游返回结果的顺序。
这里说的是调度规则；普通 PHP 函数顺序调用本身不会产生并发，实际实现还需要相应的异步执行机制.3. 分类失败与有限补试
- 失败之后，不能统一重新放回队列。不同错误，需要不同处理：
短暂网络故障：允许请求层有限重试，给网络恢复的机会。
限流：有有效恢复时间时，等待后再有限补试。
整批结果不合格：符合条件时拆成两个小批次，补试一轮。
个别帖子校验失败：先保存成功结果，只补试失败帖子。
凭据、余额、权限错误：记录具体原因，不自动反复尝试。- 例如，假设一个批次中有 24 条帖子通过校验，只有 1 条失败，就没有必要让已经成功的 24 条重新调用 AI。
补试还必须有次数限制。否则，一条始终无法处理的数据，可能持续占用资源并增加调用费用。
这里要区分两层机制：请求重试处理短暂通信问题，队列补试重新安排业务处理。 两者不是同一个重试循环。4. 熔断与保存失败保护
- 单个任务失败，可以记录原因后继续处理其他任务。但如果连续发生传输故障（请求对应的服务），就需要限制新任务的启动，避免产生更多无法确认结果的请求。
项目采用三级处理：
1. 正常情况下，最多两个批次并发。
2. 出现一次最终传输失败，降为单并发探测。
3. 连续第二次最终传输失败，暂停启动新批次。- 一批任务最终因为网络问题失败，才算一次失败，中间的重试不单独计数。 暂停后要手动继续，重启程序也不会自动恢复；继续时先一次执行一个批次。
- 任务记录保存失败，就先不启动新任务，避免任务执行了却没有记录。 已经发给 AI 的任务，可能仍会继续执行。
5. 中断恢复与结果复用
分为四种情况
本地已经保存了结果：直接继续处理这份结果，不再调用 AI。
没有结果，用户还没点“继续”：这项中断任务先不执行，等用户决定。
用户点了“继续”，而且保存了任务编号：拿编号查询原任务，不重新提交。
用户点了“继续”，但没有任务编号：重新提交，但需要先提醒可能重复执行、重复收费。- 下面用 PHP 表达这个先后关系
$taskId = $task['upstream_task_id'] ?? null;
if (!isSameInput($task)) {
showError('任务内容已变化，不能直接恢复。');
} elseif (hasUsableSavedOutput($task)) {
restoreLocally($task);
} elseif (!$userConfirmedResume) {
$message = $taskId
? '是否继续查询原任务？'
: '重新提交可能重复执行、重复收费，是否继续？';
showResumeConfirmation($message);
} elseif ($taskId) {
pollOriginalTask($taskId);
} else {
createNewTask($task);
}- 任务内容变了，就不能直接用旧结果。 比如修改了帖子正文、模型或提示词，旧结果可能已经不适用。
- 有编号但查不到原任务，就说明恢复失败。 应当告诉用户，不能悄悄创建一个新任务顶替
6. 进度与风险反馈
- 页面显示完成、未完成和失败数量，并说明失败原因、能否继续。
- 进度按任务记录统计。队列空了，不代表任务都完成了，也可能还在运行。
- 结果检查通过并保存后，才算完成。
- 刷新只查看进度，不会调用 AI。继续任务需要单独操作；可能重复收费时，先提醒用户确认。
总结
- 这些优化让队列不仅能执行任务，还能回答三个问题：
任务现在处理到了哪里？
失败后应该怎么处理？
中断后可以根据什么继续？- 解决简单队列的对应方法
| 简单队列的缺点 | 对应优化 |
|---|---|
| 没有保存任务 | 任务状态与持久化 |
| 只能逐个执行 | 批次与并发控制 |
| 没有失败处理和重试机制 | 分类失败与有限补试 |
| 没有统一的故障保护规则 | 熔断与保存失败保护 |
| 没有恢复机制 | 中断恢复与结果复用 |
| 不知道进度和失败原因 | 进度与风险反馈 |
队列负责安排执行，持久化记录保留进度，失败处理和恢复规则决定下一步动作。 最重要的是分清：没有拿到结果，不代表没有执行；恢复任务，也不一定意味着重新创建任务。
版权所有
版权归属：念宇