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

mq.type 是唯一的装配开关。没配它，MqTemplate 和 @MqListener 都不会生效。

发送

@Resource
private MqTemplate mq;

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

消息体传任意对象就行，driver 负责序列化。每条消息会自动生成一个雪花 ID 当 key，去重时会用到。

消费

方法上打个注解就行，方法参数写你期望的类型，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)。

进阶用法

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

// 去重：重复消息不会再进业务（需要 MqIdempotentStore，redis-stream 自带了一个）
@MqListener(topic = "LN_Topic", group = "g", idempotent = true)
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 存到异常表，后面人工或定时补偿
    }
}

两个 driver 不太一样

这点得说在前面：两套实现的完整度并不一致。Redis Stream 这套是照着 core 的设计意图做全的，去重、批量、失败落库、reclaim 重试都接上了；RocketMQ 那套目前只做了基础收发加原生的重试和死信，去重和 RETRY_EXHAUSTED 落库还没接进去。

能力	Redis Stream	RocketMQ
收发 / 并发消费	✅	✅
去重 idempotent	✅ 内置实现	❌ 还没接，配了也不生效
失败落库 RETRY_EXHAUSTED	✅	❌ 只会进原生 DLQ
失败落库 SINGLE	✅	⚠️ 异常会被转成 RECONSUME_LATER
批量 batchSize	✅	⚠️ 逐条回调，靠 core 内存攒批
重试	✅ PEL reclaim	✅ 原生重投
死信队列	❌ 只落库	✅ 原生 %DLQ%
顺序消息	❌	❌ 用的 Concurrently
事务 / 延迟消息	❌	❌ 没暴露
序列化	Jackson	fastjson

所以现在如果你依赖去重或者失败落库，先用 Redis Stream。RocketMQ driver 能收能发，但还没「放量」，具体见下面的 TODO。

架构

             业务代码（只见 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-redis-stream	Redis Stream 实现（XADD / 消费组 XREADGROUP / XACK / XPENDING+XCLAIM reclaim），另带一个默认的去重实现
mq-starter-rocketmq	RocketMQ 实现，基于 rocketmq-spring 编程式注册
mq-spring-boot-starter	聚合 jar，业务方引这一个就够

core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测。想看更细的图文（消费链、SPI 契约、配置速查、FAQ 都在里面），打开 docs/mq-starter-全景手册.html。

配置项

@MqListener 的属性：

属性	默认	说明
topic / group	必填	主题 / 消费组
concurrency	1	并发消费线程数，实际由 driver 兑现
batchSize	1	攒批大小，大于 1 要用 List 参数
maxRetry	3	最大重试次数
tag	*	tag 过滤
idempotent	false	是否去重
errorHandler	""	失败落库的 bean 名

Redis Stream 这边的参数（前缀 mq.redis-stream，都有默认值）：

配置	默认	说明
reclaim-min-idle	30s	消息投出去多久没 ack 才 reclaim 重投，要大于单条处理耗时
idempotent.processing-ttl	60s	「处理中」标记存多久，建议大于处理耗时、小于 reclaim-min-idle
idempotent.success-ttl	300s	「已完成」标记存多久（去重窗口）
idempotent.fail-ttl	300s	失败标记保留多久
idempotent.key-prefix	mq:idem:	去重键前缀

RocketMQ 这边直接复用 rocketmq-spring 原生的 rocketmq.* 配置。

已知限制 / TODO

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

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

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

P1：延迟消息、死信队列统一抽象、消息轨迹 / 指标、序列化可插拔（现在两边一个 Jackson 一个 fastjson）、广播消费。

P2：并发度语义统一、headers 完整透传、mq.type 错配时 fail-fast、监控告警。

构建

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）在 mvn verify 阶段跑。

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

License

Apache License 2.0，Copyright 2026 EdmundNightmare。