mq-starter

把消息队列封装成一套统一接口的 Spring Boot Starter。业务代码用 MqTemplate 发、用 @MqListener 收，底层到底走 RocketMQ 还是 Redis Stream，交给一行配置决定，切换的时候业务代码一行都不用动。

思路和 JDBC 差不多：你面向 Connection / Statement 写代码，换 MySQL 还是 Oracle 只是改驱动和连接串；这里换 MQ 只是改 mq.type。

mq:
  type: redis-stream   # 换成 rocketmq 也行，业务代码不动

@Resource MqTemplate mq;
mq.send("LN_Topic", payload);

@MqListener(topic = "LN_Topic", group = "ln_consumer")
public void onData(FrontData msg) {
    // msg 已经反序列化好了，这里只写业务
}

为什么写它

之前项目里跟 MQ 打交道，producer / consumer 的注册、ack、序列化、失败重试、去重这些样板代码散落得到处都是，真要把 RocketMQ 换成别的，消费层基本得重写一遍。

mq-starter 想做的就是把这些脏活收进 starter：业务侧只保留「发什么」和「收什么」，剩下的连接管理、线程模型、确认机制都不用管。顺手把消费侧几个常见需求（去重、批量、失败落库）也做成了注解开关。

快速开始

引入

业务方只需要引一个聚合坐标，两套 driver 都打在里面，运行时按 mq.type 二选一装配：

<dependency>
    <groupId>com.njcn</groupId>
    <artifactId>mq-spring-boot-starter</artifactId>
    <version>1.0.0-SNAPSHOT</version>
</dependency>

目前还没发到中央仓库，先在项目根目录 mvn install 装到本地仓库用。

配置

用 Redis Stream：

mq:
  type: redis-stream
spring:
  redis:
    host: localhost
    port: 6379

用 RocketMQ：

mq:
  type: rocketmq
rocketmq:
  name-server: localhost:9876
  producer:
    group: my-producer-group
spring:
  redis:             # 去重默认开启、状态存 Redis，所以 rocketmq 部署也要配 Redis；
    host: localhost  # 确实不用去重时可去掉，并配 mq.idempotent.enabled=false 全局关闭
    port: 6379

mq.type 是装配开关。注意 rocketmq driver 带 matchIfMissing=true：没配 mq.type 会默认装配 rocketmq（聚合坐标下 classpath 恒有 rocketmq），此时若 rocketmq.* 没配好通常直接启动报错；想用 redis-stream 必须显式写 mq.type: redis-stream；配了两端都不认的错误值（拼错）默认直接 fail-fast——启动抛 IllegalStateException 并附中文排查提示，不是静默不装。仅当显式 mq.startup-check.enabled=false 关掉启动校验时，才退化为两个 driver 都不装、MqTemplate 和 @MqListener 都不生效。

发送

@Resource
private MqTemplate mq;

mq.send("LN_Topic", frontData);            // topic + 消息体
mq.send("File_Topic", nodeId, fileData);   // 带路由 key，映射成 tag

消息体传任意对象就行，driver 负责序列化（两端都是 fastjson，POJO 记得留公开无参构造和 getter/setter，收发两端用同一个类型）。每条消息会自动生成一个雪花 ID 当 key，去重时会用到。

发送语义有两点要知道：

• send 是同步调用、返回 void（不回传消息 ID）：方法正常返回就代表已入队（rocketmq 是等 broker 确认的 syncSend），失败直接抛异常，业务自己 try/catch 或交给上层事务回滚。
• 消费端反序列化失败两端处理不一样：redis-stream 当"毒消息"直接 ACK 丢弃，rocketmq 会反复重试直到耗尽。改消息类字段时注意兼容，建议先自测能 round-trip。

消费

方法上打个注解就行，方法参数写你期望的类型，starter 按类型反序列化：

@Component
public class FrontDataConsumer {

    @MqListener(topic = "LN_Topic", group = "ln_consumer")
    public void onFrontData(FrontData msg) {
        // ...
    }

    // 想拿 key / tag / timestamp / headers 这些元数据，就再加一个 MqContext 参数
    @MqListener(topic = "File_Topic", group = "file_consumer")
    public void onFile(FileData msg, MqContext ctx) {
        String key = ctx.getKey();
    }
}

方法签名只认这两种：(payload) 或 (payload, MqContext)。

tag：同一个 topic 下再分小类

topic 是大频道，tag 是频道内给消息贴的标签——多个下游订同一个 topic、各自只认自己的标签，就不用为每个小类单独开 topic。发送方打标签、消费方声明要哪类，就这两步：

mq.send("File_Topic", "node-01", fileData);   // 三参 send 的第二参就是 tag

@MqListener(topic = "File_Topic", group = "file_consumer", tag = "node-01")
public void onFile(FileData msg) { }          // 只收 tag=node-01 的消息，其余不会进这个方法

• 消费端不写 tag 默认 *，整个 topic 全收；发送端用两参 send 则消息不带 tag——对端若按 tag 过滤，不带标签的消息它收不到，跨系统对接时这是个经典坑。
• rocketmq 模式：过滤发生在 broker 服务端，不匹配的消息根本不会投到消费者；tag 表达式直接透传 RocketMQ 原生语法，"A || B" 多选一也支持。
• redis-stream 模式：当前不过滤、全收（见能力表），tag 只是随消息带到消费端，可从 MqContext.getTag() 读到，需要分流的话业务自己判断——这也是切换 mq.type 时的行为翻转点之一。

进阶用法

去重、批量、失败落库都是 @MqListener 上的属性，业务方法本身不用改：

// 去重默认开启：重复消息不会再进业务（去重状态存 Redis，两个 driver 都支持，见下方说明）
// 确实不需要去重的监听器显式关掉：
@MqListener(topic = "LN_Topic", group = "g", idempotent = false)
public void onData(FrontData msg) { }

// 批量：攒够 batchSize 条一次给你，参数要用 List
@MqListener(topic = "LN_Topic", group = "g", batchSize = 50)
public void onBatch(List<FrontData> msgs) { }

// 失败落库：消费失败的消息交给你自己的 handler 落库
@MqListener(topic = "LN_Topic", group = "g", errorHandler = "myErrorHandler")
public void onData2(FrontData msg) { }

落库 handler 自己实现一个 bean：

@Component("myErrorHandler")
public class MyErrorHandler implements MqConsumeErrorHandler {
    @Override
    public void onError(MqMessage message, MqErrorIdentity identity, Exception ex) {
        // identity 区分是单次失败(SINGLE)还是重试耗尽(RETRY_EXHAUSTED)
        // 把 message 存到异常表，后面人工或定时补偿
    }
}

⚠️ 不配 errorHandler = 失败静默丢弃。未指定 errorHandler 时，单条 / 批量 / 重试耗尽的任何消费失败最终都会被 ACK、不落库、不留痕迹（ACK 终止是内建行为，与是否落库无关）。关键消费务必配上 errorHandler 落库，否则失败即丢、事后无从排查。

⚠️ 批量有几个坑：参数必须写带具体元素类型的 List<FrontData>（裸 List 启动扫描直接抛 IllegalStateException）；没有 idle flush，低流量攒不满会一直等，进程重启会丢内存里没攒满的那批；整批业务失败只回调 SINGLE 落库、不重投（所以批量务必配 errorHandler）。只适合高流量、能接受这些取舍的场景。批量方法同样可以加第二参 MqContext（拿到的是该批最后一条的元数据）。

另外 batchSize 与首参是否 List 错配时不报错：启动时打一条 WARN 日志、仍按首参决定走不走批量（batchSize=50 配单个 payload 退化为逐条；List 首参配 batchSize 缺省按 1 条一批）。见启动日志 [mq] @MqListener batchSize=...。

两个 driver 不太一样

两套实现的能力大头已经对齐：去重、失败落库两端都接上了。剩下的差异主要在死信模型和批量语义：

能力	Redis Stream	RocketMQ
收发	✅	✅
并发消费 concurrency	⚠️ 单线程 poll，未兑现	⚠️ 原生线程池，不受 concurrency 控制
去重 idempotent（默认开启）	✅（需 Redis）	✅（需 Redis）
失败落库 RETRY_EXHAUSTED（需配 errorHandler，否则仅 ACK 丢弃）	✅	✅ 落库+ACK
失败落库 SINGLE	⚠️ 单条：落库+重投；批量：失败被吞（两端一致）	⚠️ 单条：落库+重投；批量：失败被吞（两端一致）
批量 batchSize	⚠️ core 内存攒批（driver 逐条喂）；无 idle flush，低频停滞/重启丢未满批（两端一致）	⚠️ core 内存攒批（driver 逐条喂）；无 idle flush，低频停滞/重启丢未满批（两端一致）
重试	✅ PEL reclaim	✅ 原生重投
死信队列	❌ 只落库	⚠️ 原生 %DLQ% 通常被耗尽 ACK 旁路（InProgress 白跑烧光计数时仍会进）
顺序消息	❌	❌ 用的 Concurrently
事务 / 延迟消息	❌	❌ 没暴露
序列化	fastjson	fastjson

⚠️ 切换 mq.type 业务代码零改动，但三处运行时行为会跟着翻转，切换前按这三条过一遍消费方假设：

1. tag 过滤：redis-stream 消费端不过滤（全收）、rocketmq 走 broker 服务端按 tag 过滤——依赖 tag 分流的业务切 rocketmq 会"只收匹配的"、切回 redis 会"收到全部"。
2. 并发模型：redis-stream 单线程串行、rocketmq 原生线程池并发——消费逻辑若有非线程安全假设，切到 rocketmq 可能暴露。
3. 顺序：两端都不保证严格有序（rocketmq 用的 Concurrently 而非 Orderly）。

去重按消息 key 判重：经 MqTemplate 发送时自动生成雪花 ID、天然唯一，业务无感；不经 MqTemplate 的异构发送方必须自己保证 key 逐条唯一（redis-stream = 信封 key field，rocketmq = 报文头 KEYS 属性；为空或固定值会把不同消息误判成重复而静默丢弃，详见「Redis Stream 消息结构」一节及其末尾的 RocketMQ 补注）。去重的状态存在 Redis（processing / success / fail 三态 + TTL），由独立模块 mq-starter-idempotent-redis 提供，不看 mq.type——所以 RocketMQ 部署只要能连上 Redis（标准 spring.redis.* 配置），去重一样生效。引聚合坐标的话这个模块已经打在里面；单独只引 mq-starter-rocketmq 的话需要自己再加一个 mq-starter-idempotent-redis 依赖。去重默认开启，消费链路会读写 Redis，务必保证 Redis 可达——连不上时消费抛连接异常走重投（日志可见、不是静默丢，但业务会一直消费不进去）。确实不需要去重的监听器显式写 idempotent = false；或配 mq.idempotent.enabled=false 全局关掉 store（关掉后去重不生效、启动打 WARN 提示，完全不碰 Redis、不影响启动）。

Redis Stream 消息结构（异构发送方对接）

redis-stream 模式下，消息在 Redis 里不是裸 JSON，而是固定五字段的信封（XADD 的 field），契约与内部 redis-stream-springboot-starter 一致（该格式当初就是和 C++ 端对死的）。不经 mq-starter 的发送方（C++ / 脚本 / 其他语言）直接 XADD 进对应 stream，@MqListener 一样能收，只要按这个结构写：

field	内容
key	消息业务 key，每条必须唯一——消费端去重按它判重（idempotent 默认开启），key 重复或恒定会把不同消息误判成重复而丢弃。经 MqTemplate 发送时自动生成雪花 ID，无需关心
tag	路由 / 过滤标记，可为空串（注意 redis-stream 消费端当前不按 tag 过滤，全收）
enc	body 的编码方式，压缩标识就放这里：plain（明文 JSON）或 gzip（压缩，见下）
body	消息体：enc=plain 时为 JSON 明文；enc=gzip 时为 base64(gzip(json))——先标准 gzip（RFC1952，带 gzip 头），再 base64（基础编码、不带换行）
ts	发送时间戳（毫秒）

发送方需要压缩时（大报文省 Redis 内存）：enc 填 gzip，body 填「JSON → gzip → base64」的结果；不压缩就 enc 填 plain、body 放 JSON 明文。只有 body 参与压缩，其余四个字段恒明文。消费端按 enc 自动解码，两种都能收。enc 别自造取值：未知值会被兜底当明文处理，body 若实际是压缩数据就会解析失败、按毒消息 ACK 丢弃。

# 明文
XADD LN_Topic * key 1815728470893350912 tag "" enc plain body '{"id":1,"name":"..."}' ts 1752480000000
# 压缩：body 为 base64(gzip(json))
XADD LN_Topic * key 1815728470893350913 tag "" enc gzip body 'H4sIAAAAAAAA...' ts 1752480000000

两个注意点：

• stream 名：默认就是 topic；若开了中间件的环境隔离（redis-stream.env-isolation），实际 stream 名是 topic_环境后缀（如 LN_Topic_test），异构发送方要写到带后缀的那个。
• 方向差：mq-starter 自己发送恒为 plain（enc 照写、恒填 plain），压缩解码能力只在消费侧。需要 mq-starter 发送侧也压缩目前做不到，要改 driver。

RocketMQ 模式没有这层信封也没有 enc——消息体就是裸 JSON，大报文由 RocketMQ 客户端自动透明压缩（发送端超约 4KB 自动压、消费端自动解），应用层不感知。

但信封里 key 的角色在 RocketMQ 模式下由报文头 KEYS 属性承担（消费端去重取的是报文头 KEYS，不读消息体里的任何字段），异构发送方要遵守同一条契约：KEYS 逐条唯一、同一条消息重投时保持不变（发送时 setKeys(唯一id)，复用消息体里已有的唯一业务 id 即可）。注意 KEYS 在原生 RocketMQ 里只是可选的运维检索字段，不填也收发一切正常、多年无症状——但接入 mq-starter 消费后它就是去重钥匙：为空或固定值时所有消息共用一把钥匙，第一条成功后去重窗口内其余全部被判重复、静默 ACK 丢弃。发送方暂时无法保证 KEYS 唯一的，对应消费端必须显式 @MqListener(idempotent = false)。

架构

             业务代码（只见 MqTemplate / @MqListener）
                            │
     ┌──────────────────────┴──────────────────────┐
     │                mq-starter-core               │
     │  MqTemplate · @MqListener 扫描 · MqMessage    │
     │  消费链：去重 → 攒批 → 消费 → 落库/重试        │
     │  SPI：MqDriver / MqSubscription / ...         │
     └──────────────┬──────────────────┬────────────┘
        @ConditionalOnProperty(mq.type)  二选一
     ┌──────────────┴─────┐   ┌─────────┴───────────┐
     │ redis-stream driver │   │  rocketmq driver    │
     └─────────────────────┘   └─────────────────────┘

五个模块：

模块	干什么
mq-starter-core	门面 MqTemplate、@MqListener 扫描注册、中立信封 MqMessage、消费处理链、几个 SPI，还有基于 SmartLifecycle 的启停
mq-starter-idempotent-redis	消费去重的 Redis 存储（driver 中立，不看 mq.type，有 Redis 就装配；配置前缀 mq.idempotent）
mq-starter-redis-stream	Redis Stream 实现（XADD / 消费组 XREADGROUP / XACK / XPENDING+XCLAIM reclaim），去重经依赖 idempotent-redis 获得
mq-starter-rocketmq	RocketMQ 实现，基于 rocketmq-spring 编程式注册
mq-spring-boot-starter	聚合 jar，业务方引这一个就够（恒含 idempotent-redis）

core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测。想看更细的架构图文（消费链、SPI 契约、driver 能力矩阵、改进 Roadmap），打开 docs/mq-starter-全景手册.html；上手和排错看本 README 就够了。

配置项

@MqListener 的属性：

属性	默认	说明
topic	必填	订阅的主题，和发送端 send(topic, ...) 传的 topic 一致才收得到
group	必填	消费组名。同一组的多个实例分摊消费（一条消息只被组内一个实例消费），不同组各收一份全量。消费进度跟着组走：换个组名就是新组，两端都从最新消息开始收、不回放历史（rocketmq 新组默认最新位点，redis-stream 建组用 $）
concurrency	1	并发消费线程数，当前两 driver 均未按此字段兑现（redis 单线程 poll、rocketmq 用原生线程池默认值）
batchSize	1	攒批大小，大于 1 要用 List 参数；与首参是否 List 错配时启动打 WARN、仍按首参跑
maxRetry	3	最大重试次数（redis-stream 按 PEL delivery 计数判耗尽 / rocketmq 写入 broker maxReconsumeTimes；被去重标记挡回的 InProgress 白跑两端都计入）
tag	*	只收打了指定 tag 的消息（发送端三参 send 的第二参打 tag）。rocketmq 为 broker 服务端过滤、支持原生表达式（如 A || B）；redis-stream 当前不过滤、全收。详见「消费」一节的 tag 小节
idempotent	true	是否去重（默认开启，需 Redis 可达；单个监听器关闭写 idempotent = false，全局关闭配 mq.idempotent.enabled=false）
errorHandler	""	失败落库的 bean 名

去重参数（前缀 mq.idempotent，driver 中立，都有默认值）：

配置	默认	说明
enabled	true	逃生开关，显式 false 关闭去重 store 装配
processing-ttl	60s	「处理中」标记存多久，必须大于最长单条处理耗时，且两端都有上界（被标记挡回的 InProgress 白跑会烧重试计数）：redis-stream 下需 ≥ reclaim-min-idle 且 < maxRetry × reclaim-min-idle（下界防慢消费双消费；标记活过全部 reclaim 轮次则 delivery 烧光、业务没重跑就判耗尽——未配 errorHandler 即静默丢弃），默认 30 ≤ 60 < 90 自洽；rocketmq 下需小于前 maxRetry 轮 broker 重投累计时长（默认 10s+30s+1m≈100s），否则并发中 / 崩溃残留的消息把 reconsumeTimes 烧光、进原生 %DLQ%
success-ttl	300s	「已完成」标记存多久（去重窗口）
fail-ttl	300s	失败标记保留多久（fail 态不挡重投，重投后允许重新消费）
key-prefix	mq:idem:	去重键前缀

旧键 mq.redis-stream.idempotent.* 仍向后兼容（会打一条弃用告警），中立键和旧键同时配置时以 mq.idempotent.* 为准。建议尽快迁移。

Redis Stream 这边的参数（前缀 mq.redis-stream）：

配置	默认	说明
reclaim-min-idle	30s	消息投出去多久没 ack 才 reclaim 重投，要大于单条处理耗时；不要调到比 processing-ttl 还大（需保持 processing-ttl ≥ reclaim-min-idle，否则存活的慢消费者会被 reclaim 重投副本并行双消费）；也别调得太小——每次 reclaim（xClaim）都使 PEL delivery +1，被 processing 挡回的白跑同样计入，需保持 maxRetry × reclaim-min-idle > processing-ttl，否则崩溃残留会在业务没重跑的情况下烧光重试直接判耗尽

Redis 连接走标准 spring.redis.*；更底层的 redis-stream.*（readCount / blockMs / consumerName 等）来自内部 stream 中间件（redis-stream-springboot-starter），一般不用动。

RocketMQ 这边直接复用 rocketmq-spring 原生的 rocketmq.* 配置（name-server、producer.group 等）。

排错 FAQ

启动找不到 MqTemplate bean / 一启动就抛 MQ 相关异常？ 先看 mq.type：拼错了默认直接 fail-fast（IllegalStateException + 中文排查提示，检查依赖和拼写）；完全没配则默认装 rocketmq，rocketmq.* 没配好通常直接启动报错。想用 redis-stream 必须显式写 mq.type: redis-stream。详见上文「配置」一节。

@MqListener 方法不触发？ 依次查：① 所在类是不是 Spring bean；② topic / group 和发送端是否一致；③ 方法签名是否合法（单条 (payload) 或 (payload, MqContext)，批量首参 List<T> 带具体元素类型）。

去重在 rocketmq 下不生效？ 依次查：① Redis 是否可达（标准 spring.redis.* 配置）；② classpath 有没有 mq-starter-idempotent-redis（聚合坐标已含，单独引 rocketmq 模块要自己加）；③ 是否配了 mq.idempotent.enabled=false。开了去重但 Redis 连不上时，消费会抛连接异常走重投，日志可见，不是静默丢。

消息重复消费？ 去重默认开启（idempotent 默认 true），正常不该重复。依次查：① 是否显式关掉了（监听器上 idempotent = false，或全局 mq.idempotent.enabled=false）；② 去重 store 是否已装配、Redis 是否可达（默认实现随 mq-starter-idempotent-redis 自动注册，不看 mq.type）。

对接异构发送方后，只有第一条消息能消费、后面成片丢失？ 几乎必是去重 key 撞车：发送方没填或写死了 key（redis-stream 信封 key field / rocketmq 报文头 KEYS 属性），所有消息共用一把去重钥匙，第一条成功后其余全被判重复静默 ACK。让发送方保证 key 逐条唯一（见「Redis Stream 消息结构」一节），兑现前先在对应监听器上显式 idempotent = false。

批量消费收不到 / 停滞 / 丢消息？ 没有 idle flush，攒不满 batchSize 不触发；整批失败会被吞掉不重投。低流量把 batchSize 调小或改单条消费。

消费失败后消息不见了、事后查不到？ 多半是没配 errorHandler：没配时单条 / 批量 / 重试耗尽的失败一律只 ACK 丢弃、不留痕迹。关键消费用 @MqListener(errorHandler = "beanName") 落库，见「进阶用法」。

开了去重后某条消息迟迟不消费 / 反复重投？ 多半是「处理中」标记没过期挡住了重投：消费者崩溃残留标记，或单条处理耗时超过 mq.idempotent.processing-ttl。等 TTL 过期会自动恢复；调 processing-ttl 时注意配置表里的上下界约束。rocketmq 下持续重投烧光 maxReconsumeTimes 会进原生 %DLQ%（进死信，不是丢消息）。

从 redis-stream 切到 rocketmq 后行为变了（多收 / 漏收、并发不一样）？ 正常，见「两个 driver 不太一样」里的三条运行时翻转：tag 过滤、并发模型、顺序保证。

已知限制 / TODO

按优先级列，详细的在 docs/mq-starter-改进优化清单.md。

P0，放量到生产前想补的：

• 失败落库、去重、批量消费（core 已做）
• 给 rocketmq driver 补上去重和 RETRY_EXHAUSTED 落库，跟 redis-stream 对齐（去重 store 已抽成 driver 中立的 mq-starter-idempotent-redis）
• 批量消费加 idle flush，现在低流量下攒不满会卡着，进程重启还会丢内存里没攒满的那批
• 顺序消息
• 事务消息

P1：延迟消息、死信队列统一抽象、消息轨迹 / 指标、序列化可插拔（两端都是 fastjson，但硬编码换不掉）、广播消费。

P2：并发度语义统一、headers 完整透传、监控告警。（启动期 fail-fast 自检 —— mq.type 错配 / 中间件连不上时启动报错 —— 已于 2026-07-09 实现，逃生开关 mq.startup-check.enabled=false；详见全景手册 Roadmap 13。）

构建

git clone https://github.com/EdmundNightmare/mq-starter.git
cd mq-starter
mvn clean install

需要 JDK 1.8+ 和 Maven 3.6+。单元测试 mvn test 不依赖外部中间件；带真实 Redis 的集成测试（*IT，连 localhost:6379）在 mvn verify 阶段跑，测试内部会自检连接、连不上 Redis 自动跳过而非失败。

有想法欢迎提 issue 或 PR，优先认领上面 P0 的几项，提交前跑一下 mvn verify，改动尽量带上测试。

License

Apache License 2.0，Copyright 2026 EdmundNightmare。