easy-messaging 使用文档

easy-messaging 提供后端中立的消息管道——通过 MessageSender<T> / MessageListener<T> 门面屏蔽 MQ 差异，底层 SPI 支持 7 种后端（RocketMQ / Redis / Kafka / RabbitMQ / ZeroMQ / DBMS / AutoMQ）。注解驱动（@Producer / @Consumer）最简配置，编程式 SPI 灵活组合。所有后端 optional scope，按需引入；同一应用只启用一个 MQ 后端（互斥 @ConditionalOnMissingBean）。

• 1. 引入依赖
• 2. 选择 MQ 后端
• 3. Producer 侧：发送消息
• 4. Consumer 侧：消费消息
• 5. 配置详解
• 6. 编程式 SPI 用法
• 7. 与 easy-event-bus 衔接
• 8. 常见问题

1. 引入依赖

<!-- 核心依赖（必需，compile scope） -->
<dependency>
    <groupId>io.matrix</groupId>
    <artifactId>easy-messaging</artifactId>
    <version>${revision}</version>
</dependency>

<!-- 至少选一个 MQ 后端（均为 optional scope，按需引入） -->
<!-- RocketMQ（生产推荐） -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
</dependency>

<!-- 或 Redis（小项目） -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

<!-- 或 DBMS（零外部依赖，测试用） -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>

2. 选择 MQ 后端

⚠️ 同一应用只启用一个 MQ 后端——所有 Backend AutoConfiguration 都有 @ConditionalOnMissingBean(BackendProducer/Consumer)。第一个匹配完成后，其他自动跳过。

DBMS 后端建表 SQL

-- Producer 默认写这张表（可配置）
CREATE TABLE easy_messaging_bus (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    topic VARCHAR(128) NOT NULL,
    headers VARCHAR(2000),
    payload TEXT,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    status VARCHAR(32) DEFAULT 'PENDING'  -- PENDING / CONSUMED / FAILED
);

-- Consumer 默认读这张表（⚠️ 注意 Producer/Consumer 默认表名**不一致**！）
-- 必须显式配置让两者用同一张表

3. Producer 侧：发送消息

3.1 最简方式：注解驱动

// 1. 定义 Topic 配置（继承 AbstractMessageSender，加 @Producer）
@Producer(topic = "order-events", tag = "created")
public class OrderEventSender extends AbstractMessageSender<OrderEvent> {}

// 2. 注入使用
@Service
public class OrderService {

    @Autowired
    private OrderEventSender eventSender;

    public void createOrder(Order order) {
        // 同步发送
        Result result = eventSender.send(new OrderEvent(order.getId(), order.getAmount()));
        if (result.isSuccess()) {
            System.out.println("发送成功：" + result.getMessageId());
        }

        // 异步发送
        eventSender.sendAsync(new OrderEvent(...), new Callback() {
            @Override
            public void onSuccess(Result result) { ... }
            @Override
            public void onException(Throwable ex) { ... }
        });

        // OneWay（fire-and-forget）
        eventSender.sendOneWay(new OrderEvent(...));

        // 顺序发送（仅 RocketMQ 有真正 hashKey 路由）
        eventSender.sendOrderly(new OrderEvent(...), order.getId());
    }
}

3.2 带自定义 Headers

Headers headers = Headers.create("trace-id", "abc123").create("biz", "order");
eventSender.send(headers, new OrderEvent(...));

3.3 @Producer 参数

4. Consumer 侧：消费消息

4.1 最简方式：注解驱动

// 1. 定义监听器（继承 MessageListener，加 @Consumer）
@Consumer(
    topic = "order-events",
    tag = "created",
    threadMin = 8,
    threadMax = 32
)
@Component
public class OrderEventConsumer extends MessageListener<OrderEvent> {

    @Override
    public void onMessage(Headers headers, OrderEvent event) throws Exception {
        // 业务处理
        inventoryService.deduct(event.getOrderId(), event.getAmount());
        log.info("处理订单事件：{}", event.getOrderId());
    }
}
// 2. 自动注册：afterPropertiesSet() → backendConsumer.registerMessageListener() + start()

4.2 @Consumer 参数

5. 配置详解

5.1 启用后端

# 启用 RocketMQ
easy:
  rocketmq:
    enabled: true

rocketmq:
  name-server: localhost:9876

# 或启用 Redis
easy:
  redis:
    enabled: true

spring:
  redis:
    host: localhost
    port: 6379

# 或启用 DBMS（零外部依赖，需先建表）
easy:
  dbms:
    enabled: true

5.2 自动配置加载顺序

各 Backend AutoConfiguration（@AutoConfigureBefore(MessagingAutoConfiguration.class)）
    ↓ 注入 BackendProducer + BackendConsumer
MessagingAutoConfiguration（汇聚锚点）
    ↓ 注册 MessageListenerBeanPostProcessor
[easy-event-bus CqrsMessagingAutoConfiguration（@AutoConfigureAfter，类名为历史命名）]
    ↓ 桥接到事件总线门面

互斥逻辑：所有 Backend Bean 定义都带 @ConditionalOnMissingBean(BackendProducer.class) + @ConditionalOnMissingBean(BackendConsumer.class)。第一个匹配完成后，后续 Bean 全部跳过。

6. 编程式 SPI 用法

6.1 直接注入 BackendProducer

@Service
public class MyService {

    @Autowired
    private BackendProducer backendProducer;

    public void doSomething() {
        backendProducer.syncSend("my-topic", "hello");
    }
}

6.2 多后端场景（绕开互斥条件）

如果真的需要同时启用两个后端（比如 RocketMQ 发命令、Redis 发实时通知），可以：

1. 只给一个 Backend AutoConfiguration 加 @ConditionalOnProperty
2. 另一个后端的 Bean 手动创建，不用 @ConditionalOnMissingBean 保护
3. 各自命名注入

7. 与 easy-event-bus 衔接

easy-event-bus 的 Adapter 层通过 EasyMessagingProducerAdapter 把领域事件桥接到 MessageSender：

DomainEvent -> EasyMessagingProducerAdapter -> MessageSender<T> -> BackendProducer SPI -> MQ

easy-event-bus 的 CqrsMessagingAutoConfiguration（类名为历史命名，保持不变）用 @AutoConfigureAfter(MessagingAutoConfiguration.class) 保证顺序。业务层只需定义 DomainEvent 和 DomainEventHandler，不需要感知 MQ 类型——后端切换（RocketMQ → Redis → DBMS）零代码改动。

8. 常见问题

Q: 能同时用两个 MQ 后端吗？ A: 默认不行。所有 Backend Bean 都有 @ConditionalOnMissingBean，第一个匹配完成后其他全部跳过。要多后端需手动创建其中一个的 Bean 并去掉互斥条件。

Q: DBMS 后端 Producer/Consumer 默认表名为什么不一样？ A: Producer 默认 easy_messaging_bus，Consumer 默认 event_message，这是一个已知 bug。必须显式配置让两者使用同一张表。

Q: 除了 RocketMQ，其他后端的顺序发送 hashKey 被忽略？ A: 是的。sendOrderly / sendOneWayOrderly 在 Redis/Kafka/RabbitMQ/ZeroMQ/DBMS 实现里直接委托普通 send，hashKey 参数没用。只有 RocketMQ 有真正的 ProducerTemplate.syncSendOrderly(hashKey) 路由。

Q: 消息反序列化失败怎么办？ A: Consumer 侧用 Class.forName(messageType.getTypeName()) 反序列化，要求：(1) Payload 是 POJO 且有默认构造函数；(2) 类名在 Consumer classpath 可访问。

Q: 为什么 DBMS 后端表有 status 列？ A: DBMS Consumer 轮询 status='PENDING' 的消息，消费成功改为 CONSUMED，失败改为 FAILED。这是轮询式数据库消息队列的通用模式。

Q: AutoMQ 和 Kafka 有什么区别？ A: 没有区别。AutoMQ 后端完全继承 KafkaBackendProducer/Consumer，零增量代码，纯协议兼容。

Q: ConsumeMode.ORDERLY 怎么没用？ A: ConsumeConfig.toConsumeGroup() 硬编码 CONCURRENTLY，ORDERLY 是预留枚举值。