Queue Starter

受众：开发者 摘要：Laravel-like Job 投递与消费 Facade。sync / memory / redis / database / rabbitmq 五种 provider，失败重试、声明式消费者、处理中任务恢复、事件日志、Worker 心跳和后台运维入口。

何时使用

场景	建议
同步邮件 / 短信 / 通知 → 改异步发送	Queue.dispatch(new SendXxxJob(...))
业务事件触发的延迟任务（例 5 分钟后）	Queue.prepare(job).delay(Duration.ofMinutes(5)).dispatch()
需要可审计、可重试的异步任务	default-provider: database
多实例共享、轻量异步	default-provider: redis
单元测试 / 单进程诊断 / lite 兼容模式	default-provider: memory 或 sync
已有 RabbitMQ 基建	启用 rabbitmq provider
高频毫秒级任务流	不用 Queue，用专门 MQ 或流式平台
强一致事务任务	不用 Queue，用数据库事务 + Event 监听

业务代码只依赖 Queue Facade、QueueJob 和 QueueJobHandler，不直接绑定 Redis / 数据库 / 内存 / MQ 实现。

快速开始

spring:
  open:
    queue:
      default-provider: database
      worker:
        enabled: true

定义 Job + Handler + 投递：

public record SendWelcomeMailJob(Long userId) implements QueueJob {
    @Override public String name() { return "mail.welcome"; }
    @Override public Map<String, Object> payload() { return Map.of("userId", userId); }
}

@Component
@QueueConsumer(queueName = "mail", concurrency = 2, maxAttempts = 5, fixedBackoffSeconds = 30, visibilityTimeoutSeconds = 300)
public class SendWelcomeMailHandler implements QueueJobHandler {
    @Override public String name() { return "mail.welcome"; }
    @Override public void handle(QueueMessage message) {
        Long userId = ((Number) message.getPayload().get("userId")).longValue();
        // 业务 Service 调用
    }
}

// 投递
Queue.dispatch(new SendWelcomeMailJob(1L));

@QueueConsumer 是可选的开发体验增强。它不会改变业务 handler contract，只为 Worker 和 database provider 提供队列名、并发下限、最大尝试次数、重试退避和可见性超时默认值。更细粒度的单次投递仍可通过 Queue.prepare(...) 覆盖。

Facade API

// 立即投递（默认 queue + provider）
Queue.dispatch(job);

// 链式配置投递
Queue.prepare(job)
    .onQueue("mail")                    // 指定 queue 名
    .delay(Duration.ofMinutes(5))       // 延迟投递
    .maxAttempts(3)                     // 最大重试次数
    .dispatch();

// 直接投递 QueueMessage（低层 API）
Queue.push(message);

Job 名称必须与 QueueJobHandler.name() 一致。Payload 必须可 JSON 序列化（不要放 Entity、连接、文件句柄等不可持久化对象）。

链式投递对象不单独覆盖 provider。当前 provider 由 spring.open.queue.default-provider 统一决定，便于应用在 sync / memory / redis / database / rabbitmq 之间按环境切换。

源码机制骨架

Queue starter 的运行链路按 Facade、Manager、Provider、Dispatcher、Worker 分层：

• Queue 是业务侧 Facade。静态方法只取当前 Spring 容器绑定的 QueueManager，不直接读取配置、不直接触碰 Redis / Database / RabbitMQ。
• QueueManagerProvider 在自动配置完成后把 QueueManager 绑定到 Facade；测试或上下文销毁时负责清理绑定。
• DefaultQueueManager 会补齐默认队列名、校验 Job 名称，把字符串 / QueueJob / QueueMessage 统一归一为 QueueMessage，再按 default-provider 选择 provider。
• PendingQueueDispatch 只覆盖队列、延迟时间、最大尝试次数和 traceId；provider 选择仍走默认配置，避免单次投递绕开运维侧全局策略。
• QueueMessage 是 provider 之间共享的标准消息：包含消息 ID、队列名、Job 名称、payload、可执行时间、最大尝试次数、当前尝试次数、traceId 和创建时间。
• QueueProvider 是底层队列介质 SPI。内置 sync、memory、redis、database、rabbitmq 五类 provider；每个 provider 只负责投递和消费一条消息。
• DefaultQueueJobDispatcher 通过 QueueJobHandler.name() 建立 Job 处理器索引；provider 消费到消息后交给 dispatcher 执行，找不到 handler 或 handler 抛异常时返回失败结果。
• QueueConsumerRegistry 收集带 @QueueConsumer 的 handler 配置，用于补齐默认队列、Worker 并发下限、最大尝试次数、重试退避和可见性超时。
• QueueWorker 是可选自动消费器。它合并显式配置队列和声明式消费者队列，按轮询间隔、并发数和每轮最大数量调用 QueueManager.work(...)，失败后触发日志和可选 QueueFailureNotifier；存在 QueueWorkerHeartbeatStore 时会记录启动、轮询、异常和停止心跳。
• database provider 负责 queue_job / queue_failed_job / queue_job_event_log / queue_worker_heartbeat 持久化：成功消费删除待执行任务；失败未达最大次数时重新置回 READY 并延迟重试；达到最大次数后写入失败任务表；任务长时间停在处理中时按处理超时配置自动恢复为 READY；关键状态变化写入事件日志，Worker 运行快照写入心跳表。
• spring-open-core-queue 提供队列指标、运维快照、处理中任务恢复、事件日志、失败任务详情、重试和删除等运维 API；Queue starter 本身只承载底层队列能力。

配置项

spring:
  open:
    queue:
      enabled: true
      default-provider: database
      default-queue-name: default
      providers:
        memory:
          provider: memory
          capacity: 10000
        redis:
          provider: redis
          enabled: true
          key-prefix: queue:jobs
        database:
          provider: database
          enabled: true
          instance-id: default
          retry-delay: 30s
          processing-timeout: 5m
          max-error-message-length: 2000
        rabbitmq:
          provider: rabbitmq
          enabled: false
          exchange: ""
          routing-key-prefix: ""
      worker:
        enabled: true
        auto-startup: true
        queues: default
        polling-interval: 1s
        concurrency: 1
        max-messages-per-poll: 10
        shutdown-timeout: 30s
        failure-notification:
          enabled: true
          channel: log

配置	默认值	说明
enabled	true	启用开关
default-provider	database	默认 provider 配置名；生产默认使用可持久化、可重试、可审计的数据库队列
default-queue-name	default	默认 queue 名
providers.<name>.provider	-	真实 provider
worker.enabled	starter 裸用 false / 应用装配 true	是否自动消费
worker.queues	default	消费的 queue 白名单（逗号分隔）
worker.concurrency	1	并发线程数
worker.max-messages-per-poll	10	每轮最大消费数
worker.polling-interval	1s	轮询间隔
worker.shutdown-timeout	30s	优雅停机等待时间
providers.database.processing-timeout	5m	处理中任务超过该时间会重新进入 READY

开发期不保留旧配置兼容；Queue 统一使用 default-provider / providers.*。

Provider

Provider 配置	Provider	说明	适用
sync	sync	同步执行，不进队列	单元测试、显式诊断场景
memory	memory	本机内存	单测、隔离诊断、lite 兼容模式
redis	redis	Redis List	多实例共享、轻量异步
database	database	数据库持久化	默认，生产稳定、可审计、可重试
rabbitmq	rabbitmq	RabbitMQ	已有 MQ 基建、业务异步解耦

database provider 使用 queue_job、queue_failed_job、queue_job_event_log 和 queue_worker_heartbeat 表。

数据库任务处理超时恢复

数据库队列消费时会先把任务从 READY 标记为 PROCESSING，并记录 reserved_at 和 locked_by。如果节点宕机、进程退出或处理器卡住，任务可能长期停留在 PROCESSING。自动 Worker 启动和每轮消费前会按 providers.database.processing-timeout 或 @QueueConsumer.visibilityTimeoutSeconds 把超时任务恢复为 READY，并在下一轮重新处理。后台 Queue 页面也提供一键恢复入口，适合人工排障后立即回收僵尸任务。

生产建议：

• processing-timeout 应大于该队列正常任务的最长处理时间，避免同一任务被重复执行。
• 需要可靠审计和失败重试时优先使用 database provider；任务处理器仍需保持幂等。
• 后台 Queue 指标会显示“超时处理中”数量，用于排查超时配置过短、处理器卡住或节点异常退出。

事件日志

database provider 会在关键节点写入 queue_job_event_log，便于后台排障和审计：

事件	说明
ENQUEUE	Job 入队或失败任务重试回队列
START	Worker 成功抢占并开始处理
SUCCESS	Handler 执行成功，待执行任务删除
RETRY	Handler 失败但未耗尽最大尝试次数，任务延迟重试
DLQ	Handler 失败且达到最大尝试次数，进入失败任务表
RECOVER	PROCESSING 超时任务被恢复为 READY

事件日志只记录运行事实和错误摘要，不替代业务审计。写入事件日志和失败任务表前，Queue 会对错误摘要中的 password、secret、apiKey、token、authorization 等 key-value 片段做脱敏；任务 payload 仍按原始业务 消息保存，敏感 payload 字段应由业务侧自行控制，不应依赖 Queue starter 做业务字段级脱敏。

后台 Queue 概览会展示最近 7 天事件趋势。趋势直接从 queue_job_event_log 聚合每日事件总数、入队、开始、成功、重试、DLQ 和恢复数量，用于判断 Worker 是否存在波动、重试堆积或死信抬升；它是运行日志视角，不替代业务订单、账务、审批或外部履约状态。

运维快照

后台 Queue 概览会通过 GET /api/v1/admin/queue/operations 展示当前队列运行快照，包括默认 provider、默认队列、Worker 开关、Worker 是否运行、监听队列、并发数、每轮最大消费数、轮询间隔、数据库处理超时、数据库指标可用性、Worker 心跳可用性、待执行 / 超时处理中 / 失败任务数量。

接口同时返回能力清单和 runbook 提示：例如 Worker 已关闭、配置开启但未运行、存在超时 PROCESSING 任务、存在失败任务、数据库指标不可用或当前运行状态健康。该快照只读，不会恢复任务、不触发重试、不修改 provider 配置，适合发布前 smoke、线上排障和开发者确认 Queue starter 是否按预期装配。

Worker 心跳

启用 database provider 时，自动 Worker 会把启动、每轮轮询、异常和停止状态写入 queue_worker_heartbeat。后台 Queue 概览通过同一个运行指标接口展示 Worker 标识、Provider、监听队列、最近心跳、最近轮询、最近一次处理 / 恢复数量、累计处理数量和最近异常。

心跳用于多节点排障：例如确认某个实例是否仍在运行、Worker 是否长时间没有 poll、最近一次是否连续失败。它不替代业务任务事件日志，也不代表任务成功率；任务生命周期仍以 queue_job_event_log 和失败任务表为准。没有 database provider 或没有心跳存储 Bean 时，Worker 仍可正常消费，只是后台会显示心跳不可用。

RabbitMQ 接入：默认关闭，且只在项目额外引入 Spring AMQP 且存在 RabbitTemplate Bean 时装配：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring:
  open:
    queue:
      default-provider: rabbitmq
      providers:
        rabbitmq:
          enabled: true
          exchange: ""
          routing-key-prefix: ""

第一阶段 RabbitMQ provider 使用 RabbitTemplate 做轻量投递和轮询消费。生产如需更强的延迟队列、死信队列、ACK 策略或独立扩缩容，建议后续单独增强或使用专用 MQ 客户端。

自动 Worker

spring.open.queue.worker.enabled 控制自动 worker：

• starter 裸用默认关闭（避免引入 starter 后意外消费）
• spring-open-application 默认开启，可通过 SPRING_OPEN_QUEUE_WORKER_ENABLED=false 关闭

Worker 通过统一 QueueManager.work(...) 消费，不直接绑定具体 provider API。

带 @QueueConsumer 的 handler 会自动加入 Worker 关注队列；如果注解声明的并发数大于全局 worker.concurrency，Worker 会采用更高并发下限。生产仍建议按整体资源容量显式配置 Worker 并发，避免单个高并发 handler 挤占数据库连接或外部服务额度。

失败通知

Worker 失败时通过 QueueFailureNotifier 扩展点通知。应用装配默认走 log 通道；启用 Notification starter 后可切换到 mail / sms / webhook / database 通道。

扩展边界

• 新增底层队列介质时实现 QueueProvider，并在自动配置中注册 provider Bean；业务代码仍只使用 Queue / QueueJob / QueueJobHandler。
• 新增业务任务时实现 QueueJob 和同名 QueueJobHandler，payload 只放可序列化字段，不放 Entity、连接、文件句柄或临时上下文对象。
• 需要 handler 级默认队列、最大尝试次数、退避或可见性超时时，优先使用 @QueueConsumer；需要单次差异化投递时使用 Queue.prepare(...)。
• 需要定时触发任务时使用 Scheduler 产生投递动作，Queue 只负责“投递后如何排队和消费”。
• 需要本进程内同步事件传播时使用 Event；需要跨实例、可审计、可重试的异步执行时使用 Queue。
• Queue 不承担复杂工作流状态机、长事务编排、高频流式消费或强一致事务保障；这些场景应使用业务 Service、审批 / 工作流底座或专用 MQ / 流式平台。

后台运维 API

spring-open-core-queue 提供队列运维入口（权限 queue:*）：

API	说明
GET /api/v1/admin/queue/metrics	队列指标（待执行 / 延迟 / 处理中 / 超时处理中 / 重试中 / 失败任务数量）和 Worker 心跳摘要
GET /api/v1/admin/queue/operations	Queue 运维快照、能力清单和 runbook 提示
POST /api/v1/admin/queue/recover	恢复超时 PROCESSING 任务
GET /api/v1/admin/queue/event-logs	Job 事件日志分页（入队 / 开始 / 成功 / 重试 / 失败入死信 / 恢复）
GET /api/v1/admin/queue/event-logs/trends	Job 事件日志每日趋势（默认最近 7 天，最长 90 天）
GET /api/v1/admin/queue/failed-jobs	失败任务分页
GET /api/v1/admin/queue/failed-jobs/{id}	失败任务详情
POST /api/v1/admin/queue/failed-jobs/{id}/retry	重试失败任务
DELETE /api/v1/admin/queue/failed-jobs/{id}	删除失败任务
DELETE /api/v1/admin/queue/failed-jobs	批量删除失败任务

注意事项

• Job payload 必须可 JSON 序列化（不放 Entity、连接、文件句柄）
• Job 名称必须与 QueueJobHandler.name() 一致
• 多实例部署优先 database 或 redis provider
• 需要可靠审计和失败重试时使用 database
• 使用 database provider 时按任务最长执行时间调整 providers.database.processing-timeout
• 不在 worker 处理器内做长事务；事务边界在被调用的 Service
• Job 应幂等：worker 可能因失败重试而执行多次

验证命令

./mvnw -pl spring-open-starters/spring-open-starter-queue -am test -DskipITs
./mvnw -pl spring-open-core/spring-open-core-queue -am test -DskipITs -DskipFrontend -Dsurefire.failIfNoSpecifiedTests=false
./mvnw -pl spring-open-application -am test -DskipITs -DskipFrontend

相关

• Scheduler（定时调度）：scheduler.md
• Event（同步 / 异步事件）：event.md
• Pipeline（责任链编排）：pipeline.md
• 核心模块治理：module-governance.md
• Provider 扩展模型：provider.md