企业官网建设流程全解析

1. 背景

本文基于一个真实的论文 AI 项目，使用 Spring Boot、MySQL 和 Worker 实现一套可恢复的异步任务调度机制，完整覆盖

需要提前说明：

MySQL 任务队列不是 MQ 的完全替代品，它适合任务量可控、需要直接查询任务状态、希望降低基础设施复杂度的中小型系统。

​​​​​​​例如，用户点击“生成论文大纲”后，系统可能需要完成：

• 调用 AI 模型；
• 解析模型返回内容；
• 保存章节任务；
• 保存对话记录；
• 保存 RAG 检索证据；
• 更新论文项目状态。

如果接口一直等待 AI 返回结果，容易出现请求超时。更严重的是，如果服务在执行过程中重启，任务状态可能丢失，用户也不知道任务到底执行到了哪一步。

常见方案是引入 RabbitMQ、Kafka 等消息队列。但对于中小型项目来说，单独部署 MQ 会增加系统复杂度。

本文介绍一种不依赖 MQ的实现方式：

使用 MySQL 保存任务，使用 Spring 定时任务轮询数据库，通过条件更新实现任务抢占，再配合心跳、超时恢复和指数退避完成可靠调度。

本实例来自一个AI项目的实现类：

二. 整体执行流程：

以“生成论文大纲”为例，任务执行流程如下：

系统中主要有两条接口：

POST /agent/thesis/project/{projectId}/outline/trigger

这个接口只负责创建异步任务，立即返回用户ID

还有同步接口：

POST /agent/thesis/project/{projectId}/outline/generate

同步接口会直接执行 AI 逻辑，因此，项目当前是“同步接口和异步队列并存”，不能简单说成所有任务都由 Worker 执行

3. 为什么不直接使用@Async

这种方式能够异步执行，但它有几个明显问题：

3.1 服务重启后任务状态丢失

任务只存在于线程池中，应用重启以后，线程池里的任务就消失了

3.2 无法查询任务进度

用户无法知道任务是等待执行、正在执行，还是已经失败

3.3 多实例部署容易重复执行

如果部署了两个服务实例，它们都可能执行同一个任务

3.4 没有统一的失败恢复机制

任务失败后，通常只能依赖代码内部处理，难以统一实现超时回收和重试

因此本项目使用了更持久化的方式：

接口请求只负责数据库，worker负责数据库的读取并行

4. 任务表设计

项目使用thesis_run_task表保存论文 AI 执行任务

任务状态定义如下：

任务类型包括：

这些字段分别解决了不同的问题：

图片中提到的heartbeat_at在本项目中并不存在，项目使用updated_at来承担心跳时间的作用

5. 创建异步任务

当用户调用异步接口时，系统不会立即执行 AI，而是先创建一条数据库任务

代码如下：

6. Worker 如何轮询数据库：

Worker 使用 Spring 的@Scheduled定时执行轮询：

含义是：

查询就绪任务的 SQL：

其中：

表示异步消费任务

7. 多个 Worker 如何避免重复执行

假设系统部署了两个实例：

两个 Worker 可能同时查询到任务5001。

如果代码只是先查询再执行，就可能出现：

这样同一个 AI 任务就被执行了两次。

项目使用带条件的UPDATE进行任务认领：

然后根据 SQL 影响行数判断是否认领成功：

执行结果如下：

原因是 Worker A 已经把任务状态从PENDING改成了RUNNING，Worker B 的条件：

已经不满足。

这种方式的关键不在于最开始的SELECT，而在于后续的条件更新

8. 执行成功后的状态更新

任务认领成功后，Worker 执行具体业务：

业务执行成功后，更新任务状态：

再次检查：

可以防止已经被其他逻辑取消或回收的任务被错误地写成成功。

9. 失败任务如何进行指数退避

AI 调用失败时，并不是所有异常都应该立即失败。

例如以下错误通常适合重试：

worker判断是否可以重试：

重试时间使用指数退避：

当基础时间为 5 秒时：

10. Mysql和 MQ 方案的对比：

对于当前论文 AI 项目来说，任务量并不是无限大的，而且后台需要直接查看任务状态，因此使用 MySQL 任务表比较合适。

如果未来任务量明显增长，或者需要跨多个服务分发任务，再考虑引入 MQ 更合理。

11.线程池拒绝也必须处理

当前项目流程是：

先领任务再提交线程

如果线程池满了，taskExecutor.execute()可能抛出拒绝异常

此时数据库任务已经是RUNNING，但任务实际还没有开始执行.如果不处理，只能等超时恢复

生产环境建议捕获线程池拒绝：

对应SQL:

任务状态覆盖完整闭环：

认领成功 提交成功 提交失败 执行成功 执行失败 执行超时

11. 失败后如何重试

并不是所有异常都应该立即失败。

项目当前识别的可重试错误包括：

overloaded rate_limit 429 timeout 超时 执行失败 未生成有效内容

代码判断：

private boolean isRetryable(RuntimeException ex) { String message = errorText(ex); return message.contains("overloaded") || message.contains("rate_limit") || message.contains("429") || message.contains("timeout") || message.contains("超时") || message.contains("执行失败") || message.contains("未生成有效"); }

11.1 指数退避

当前项目的基础退避逻辑：

private long retryDelaySeconds(Integer attemptCount) { int exponent = Math.max( 0, Math.min( 6, (attemptCount == null ? 1 : attemptCount) - 1 ) ); return Math.min( 300, Math.max(1, retryBaseSeconds) * (1L << exponent) ); }

基础时间为 5 秒时：

第一次失败：5 秒后重试 第二次失败：10 秒后重试 第三次失败：20 秒后重试 超过最大次数：FAILED

11.2 增加随机抖动

多个任务同时失败时，如果都在固定的 5 秒、10 秒后重试，可能再次形成请求洪峰。

可以增加随机抖动：

private long retryDelaySeconds( int attempt, long baseSeconds, long maxSeconds) { int exponent = Math.min( Math.max(attempt - 1, 0), 6 ); long exponential = baseSeconds * (1L << exponent); long capped = Math.min(exponential, maxSeconds); long jitter = ThreadLocalRandom.current().nextLong( Math.max(1, capped / 5) + 1 ); return capped + jitter; }

调度重试：

UPDATE thesis_run_task SET status = 'RETRYING', worker_id = NULL, next_retry_at = #{nextRetryAt}, error_code = #{errorCode}, error_message = #{errorMessage}, updated_at = NOW() WHERE id = #{taskId} AND status = 'RUNNING' AND worker_id = #{workerId} AND attempt_count < max_attempts;

项目中还存在一层 AI 调用内部重试：

AI 单次调用内部：2 秒、4 秒 队列任务外层：5 秒、10 秒、20 秒

这两层重试需要分别配置，不能把它们混成一个重试机制。

12. 心跳和超时恢复

12.1 心跳

Worker 执行任务期间会定时更新updated_at：

ScheduledFuture<?> heartbeat = heartbeatScheduler.scheduleAtFixedRate( () -> sendHeartbeat( taskId, claimWorkerId ), Math.max(1, heartbeatSeconds), Math.max(1, heartbeatSeconds), TimeUnit.SECONDS );

SQL：

UPDATE thesis_run_task SET updated_at = NOW() WHERE id = #{taskId} AND status = 'RUNNING' AND worker_id = #{workerId};

12.2 超时回滚

Worker 定时扫描超时任务：

@Scheduled( fixedDelayString = "${thesis.task.worker.recovery-delay-ms:60000}" ) public void recoverTimedOutTasks() { LocalDateTime cutoff = LocalDateTime.now() .minusSeconds( Math.max(60, timeoutSeconds) ); int failed = thesisRunTaskMapper .failTimedOutTasks(cutoff); int requeued = thesisRunTaskMapper .requeueTimedOutTasks(cutoff); if (failed > 0 || requeued > 0) { log.warn( "Recovered timed out thesis tasks: " + "requeued={}, failed={}", requeued, failed ); } }

未达到最大次数的任务重新入队：

UPDATE thesis_run_task SET status = 'RETRYING', worker_id = NULL, next_retry_at = NOW(), error_code = 'WORKER_TIMEOUT', error_message = 'Worker heartbeat timed out and task was requeued', updated_at = NOW() WHERE status = 'RUNNING' AND updated_at < #{cutoff} AND idempotency_key LIKE 'queue:%' AND attempt_count < max_attempts;

已达到最大次数的任务标记失败：

UPDATE thesis_run_task SET status = 'FAILED', worker_id = NULL, finished_at = NOW(), error_code = 'WORKER_TIMEOUT', error_message = 'Worker heartbeat timed out and retries were exhausted', updated_at = NOW() WHERE status = 'RUNNING' AND updated_at < #{cutoff} AND idempotency_key LIKE 'queue:%' AND attempt_count >= max_attempts;

状态闭环如下：

RUNNING | +-- 心跳正常 --> 继续执行 | +-- 心跳超时且未耗尽次数 --> RETRYING | +-- 心跳超时且达到最大次数 --> FAILED

生产环境可以把updated_at改为独立的lease_until字段，让业务更新时间和租约时间分离。

13. 取消任务的边界

项目提供取消接口：

POST /agent/thesis/project/run-task/{runTaskId}/cancel

取消 SQL：

UPDATE thesis_run_task SET status = 'CANCELLED', cancel_flag = 'CANCEL', worker_id = NULL, finished_at = NOW(), updated_at = NOW() WHERE id = #{taskId} AND status IN ( 'PENDING', 'RUNNING', 'RETRYING', 'PAUSED' );

取消后，Worker 不会再次领取这个任务。

但当前实现不能保证正在执行的 AI 请求立即中断：

未开始执行的任务：可以取消 等待重试的任务：可以取消 已经执行的任务：父任务可变为 CANCELLED， 但外部 AI 请求可能继续执行

因此，文章不能写成“支持实时取消”，准确说法应该是：

项目支持任务状态取消，但正在执行的外部 AI 请求不一定能够立即终止。

14. Redis 预筛选如何减少数据库压力

当前项目中的 Redis 主要用于防止重复触发：

String lockKey = "thesis:trigger:lock:" + projectId + ":" + step;

它不是任务队列本身。

如果任务量增加，可以使用 Redis Sorted Set 保存就绪任务索引：

Key: task:ready Score: next_retry_at 时间戳 Member: task_id

任务入库后写入 Redis：

redisTemplate.opsForZSet().add( "task:ready", String.valueOf(task.getId()), nextRetryTimestamp );

Worker 先从 Redis 获取候选任务：

Set<String> taskIds = redisTemplate.opsForZSet() .rangeByScore( "task:ready", 0, System.currentTimeMillis(), 0, batchSize );

但 Redis 只能做预筛选，最终认领仍然必须依赖 MySQL：

Redis 找到任务 ID | v MySQL CAS 条件更新 | +--> 成功：执行任务 | +--> 失败：说明已被其他 Worker 抢走

不能把 Redis 当成唯一任务来源，否则会出现：

任务从 Redis 删除 Worker 尚未更新数据库就宕机 任务永久丢失

因此还需要定期从 MySQL 补偿扫描：

SELECT id FROM thesis_run_task WHERE status IN ('PENDING', 'RETRYING') AND enabled = 1 AND ( next_retry_at IS NULL OR next_retry_at <= NOW() ) ORDER BY created_at LIMIT 1000;

结论：

Redis 可以减少无效轮询，但 MySQL 仍然是任务状态的最终事实来源。

15. 和 MQ 方案的对比

MySQL 任务表适合：

每秒几十个以内的长任务 任务执行时间较长 需要后台查看任务详情 系统已经使用 MySQL

MQ 更适合：

每秒数千条短消息 跨多个服务分发 需要更低的投递延迟 消息生产和消费完全解耦

16. MySQL 任务队列的性能边界

不能直接说“MySQL 任务队列支持多少 QPS”，因为实际吞吐取决于：

Worker 实例数量 线程池并发数 任务平均执行时间 数据库性能 AI 服务响应速度 心跳频率 重试比例 索引命中情况 连接池大小

可以用下面的公式估算：

任务吞吐 ≈ min( Worker数量 × 单实例并发数 / 平均任务耗时, Worker数量 × 批量大小 / 轮询间隔 )

当前项目配置：

Worker 数量：1 单实例并发数：4 批量大小：4 轮询间隔：1 秒 平均任务耗时：60 秒

线程池理论吞吐：

1 × 4 / 60 ≈ 0.067 个任务/秒 ≈ 4 个任务/分钟

虽然 Worker 每秒最多可以查出 4 条任务，但这不代表每秒能完成 4 条任务，真正瓶颈是线程池并发数和任务执行时间。

如果每个任务 60 秒完成，心跳间隔 20 秒，则一个成功任务大约需要：

1 次 claim 3 次 heartbeat 1 次 complete

如果每分钟完成 4 个任务，状态写入约为：

4 × 5 = 20 次/分钟

还没有计算：

重试 任务竞争 超时扫描 业务表写入

所以，以上只是容量估算，不是压测结果。

上线前应重点监控：

claim SQL P95/P99 数据库锁等待 数据库 CPU 数据库连接池使用率 任务完成吞吐 任务积压数量 最老任务等待时间 失败率和重试率

17. 海量任务下的归档和分库分表

当任务量增长到每天数百万条时，在线任务表会不断膨胀。

17.1 增加组合索引

可以增加：

CREATE INDEX idx_task_ready ON thesis_run_task ( status, enabled, next_retry_at, created_at, id );

如果任务类型长期依赖：

idempotency_key LIKE 'queue:%'

建议增加独立字段：

source_type ENUM('QUEUE', 'INLINE') NOT NULL

然后改成：

WHERE source_type = 'QUEUE'

17.2 归档终态任务

以下任务可以定期归档：

SUCCESS FAILED CANCELLED

在线表只保留最近 90 天数据，历史任务迁移到归档表。

17.3 按时间分区

可以按created_at使用 RANGE 分区：

PARTITION BY RANGE (TO_DAYS(created_at)) ( PARTITION p202610 VALUES LESS THAN (TO_DAYS('2026-11-01')), PARTITION p202611 VALUES LESS THAN (TO_DAYS('2026-12-01')), PARTITION pmax VALUES LESS THAN MAXVALUE );

时间分区主要解决历史查询和清理问题，不会自动解决 Worker 抢占并发问题。

17.4 按项目分片

如果不同项目之间相互独立，可以使用：

project_id % 分片数量

进行分库：

project_id % 4 = 0 -> task_db_0 project_id % 4 = 1 -> task_db_1 project_id % 4 = 2 -> task_db_2 project_id % 4 = 3 -> task_db_3

分片后需要：

每个分片独立轮询 每个分片独立监控 每个分片单独限流 每个分片单独执行超时恢复

如果业务要求全局严格按照创建时间执行，分库分表会增加很多复杂度。

18. 任务系统必须具备可观测性

没有监控的异步系统，出问题时只能人工查数据库。

18.1 统计任务状态

SELECT status, task_type, COUNT(*) AS total FROM thesis_run_task GROUP BY status, task_type;

重点监控：

PENDING 数量 RETRYING 数量 RUNNING 数量 FAILED 数量

18.2 统计最老任务等待时间

SELECT TIMESTAMPDIFF( SECOND, MIN(created_at), NOW() ) AS oldest_pending_seconds FROM thesis_run_task WHERE status IN ('PENDING', 'RETRYING') AND enabled = 1;

“最老任务等待时间”通常比任务数量更有价值。

例如：

积压 100 条，但只等待 2 秒，问题不大 积压 10 条，但最老任务等待 30 分钟，需要告警

18.3 统计失败率

SELECT COUNT( CASE WHEN status = 'FAILED' THEN 1 END ) / COUNT(*) AS failure_rate FROM thesis_run_task WHERE created_at >= NOW() - INTERVAL 5 MINUTE;

18.4 使用 Micrometer 记录指标

Counter.builder("thesis_task_failed_total") .tag("task_type", taskType.name()) .tag("error_code", errorCode) .register(meterRegistry) .increment();

建议记录：

task_created_total task_claim_success_total task_claim_conflict_total task_success_total task_failed_total task_retry_total task_timeout_total task_duration_seconds task_pending_count task_oldest_pending_seconds

18.5 Worker 健康度

每个 Worker 应记录：

worker_id 最后一次轮询时间 最后一次成功认领时间 当前执行任务数 线程池活跃线程数 线程池队列长度 最后一次心跳成功时间 数据库连接状态

典型告警：

最老待执行任务超过 5 分钟 失败率超过 10% 超时回收任务持续增长 所有 Worker 心跳中断 线程池队列持续满载 数据库连接池使用率超过 80% claim SQL P95 超过阈值

19. 这套方案是否保证只执行一次

不能。

数据库条件更新可以保证：

同一时刻只有一个 Worker 成功认领同一条任务

但不能保证：

业务副作用只发生一次

例如：

1. Worker 成功调用 AI 2. AI 结果已经写入章节表 3. Worker 尚未更新任务 SUCCESS 4. Worker 进程突然崩溃 5. 超时恢复重新把任务放回 RETRYING 6. 新 Worker 再次执行

最终可能出现：

AI 被调用两次 章节数据被写入两次 外部接口被调用两次

因此，业务层还需要：

稳定业务幂等键 章节级唯一约束 外部接口幂等请求号 结果覆盖策略 断点续跑 步骤级状态记录

例如章节生成可以使用：

project_id + chapter_no + version_no

作为业务唯一键。

需要准确区分：

防止并发重复认领 不等于保证 Exactly Once

20. 排查重复调度注解

排查 Worker 时，还需要检查同一个方法是否重复声明@Scheduled。

正确写法：

@Scheduled( fixedDelayString = "${thesis.task.worker.poll-delay-ms:1000}" ) public void poll() { // 查询和认领任务 }

超时恢复是另一条独立调度：

@Scheduled( fixedDelayString = "${thesis.task.worker.recovery-delay-ms:60000}" ) public void recoverTimedOutTasks() { // 回收超时任务 }

这两个调度职责不同：

poll()：消费待执行任务 recoverTimedOutTasks()：回收卡死任务

如果同一个poll()方法重复添加相同的@Scheduled，轮询频率会被额外放大，数据库查询和任务竞争也会增加。

21. 总结

本项目通过 MySQL 实现了一套不依赖 MQ 的异步任务调度系统：

任务落库 ↓ 数据库轮询 ↓ 条件更新抢占 ↓ 线程池执行 ↓ 心跳保活 ↓ 指数退避重试 ↓ 超时回滚 ↓ 指标监控

这套方案适合：

任务量中等 任务执行时间较长 已有 MySQL 需要直接查询任务状态 暂时不希望引入 MQ

但它不适合：

每秒数千条短消息 强实时低延迟场景 跨多个服务的大规模消息分发 必须严格 Exactly Once 的业务

真正可靠的 MySQL 任务队列，不是简单地执行：

SELECT * FROM task WHERE status = 0;

而是需要完整处理：

任务抢占 租约续期 线程池拒绝 执行成功 执行失败 指数退避 超时回滚 业务幂等 任务监控 历史归档

对于当前论文 AI 项目，MySQL 任务表能够满足中小规模长任务调度需求。随着任务量增长，可以逐步增加：

Redis 就绪任务预筛选 任务归档 时间分区 分库分表 Micrometer 指标 统一告警

当数据库已经成为明显瓶颈，或者任务需要跨多个服务高吞吐分发时，再评估引入 MQ。