easy-messaging 消息总线组件设计

本文档描述 easy-messaging 组件的设计与实现。该组件提供后端中立的消息发送/消费抽象层，通过 SPI 接口（BackendProducer / BackendConsumer）支持 7 种 MQ 后端（RocketMQ / Redis / Kafka / RabbitMQ / ZeroMQ / DBMS / AutoMQ），上层门面 MessageSender<T> / MessageListener<T> 屏蔽后端差异。所有后端均为 optional scope，应用按需引入；自动配置通过 @AutoConfigureBefore(MessagingAutoConfiguration.class) 汇聚锚点模式保证顺序。easy-event-bus Adapter 层通过 EasyMessagingProducerAdapter 桥接到本组件。

• 一、背景与问题
• 二、组件定位与边界
• 三、总体架构
• 四、核心 API 草图
• 五、SPI 层与 7 后端实现
• 六、自动配置体系
• 七、消息序列化与反序列化
• 八、已知限制
• 九、关键设计决策

一、背景与问题

1.1 为什么要独立消息抽象层

Matrix Framework 需要跨进程通信能力（领域事件、网关设备下行、实时通知等），但不同项目对 MQ 的选择不同：

如果业务层硬编码 MQ 客户端（DefaultMQProducer / KafkaProducer），切换成本极高。easy-messaging 通过 SPI 把 MQ 差异隔离在 Connector 层。

1.2 与 Spring Cloud Stream 的区别

easy-messaging 选择轻量 SPI 路线，因为 Matrix Framework 已经在事件总线层提供事件语义（CQS 范式），messaging 层只需做”管道”。

二、组件定位与边界

定位：后端中立的消息管道。业务层通过 MessageSender<T> 发送、MessageListener<T> 消费，后端 SPI 自动注入唯一可用的 MQ Connector。

边界：

约束：

• 同一应用只应启用一个 MQ 后端——所有 Backend AutoConfiguration 都有 @ConditionalOnMissingBean(BackendProducer.class) + @ConditionalOnMissingBean(BackendConsumer.class)
• 只有 RocketMQ 有真正的顺序发送——其余 6 个后端 sendOrderly/sendOneWayOrderly 的 hashKey 被忽略
• DBMS Connector Producer/Consumer 默认表名不一致——Producer 默认 easy_messaging_bus，Consumer 默认 event_message，需显式配置统一
• Kafka Consumer group.id 硬编码为 "easy-messaging-consumer-group"
• 所有 MQ 后端均为 optional scope

三、总体架构

flowchart TD
    subgraph "用户层"
        A1["MessageSender<T> (接口)"]
        A2["AbstractMessageSender<T> (薄代理)"]
        A3["MessageListener<T> (抽象类)"]
        A4["@Producer / @Consumer 注解"]
        A5["Callback / Headers / Result / ConsumeMode"]
    end

    subgraph "SPI 层"
        B1["BackendProducer (12 + 1 方法)"]
        B2["BackendConsumer (4 方法)"]
    end

    subgraph "Connector 层（7 后端，互斥）"
        C1["RocketMQ<br/>原生 Message + ProducerTemplate"]
        C2["Redis<br/>Pub/Sub + JSON wrapper"]
        C3["Kafka<br/>ProducerRecord + JSON wrapper"]
        C4["RabbitMQ<br/>convertAndSend + JSON wrapper"]
        C5["ZeroMQ<br/>PUB/SUB socket + pipe 分隔 frame"]
        C6["DBMS<br/>INSERT + SELECT轮询"]
        C7["AutoMQ<br/>继承 Kafka"]
    end

    subgraph "自动配置层"
        D1["各 Backend AutoConfiguration<br/>@AutoConfigureBefore(MessagingAutoConfiguration.class)<br/>+ @ConditionalOnMissingBean 互斥"]
        D2["MessagingAutoConfiguration<br/>汇聚锚点<br/>注册 MessageListenerBeanPostProcessor"]
    end

    A2 --> B1
    A3 --> B2
    D1 -->|"注入 BackendProducer / BackendConsumer"| B1
    D1 --> B2
    B1 --> C1 & C2 & C3 & C4 & C5 & C6 & C7
    B2 --> C1 & C2 & C3 & C4 & C5 & C6 & C7
    D1 --> D2

消息格式差异：

• RocketMQ：用原生 org.apache.rocketmq.common.message.Message，不做 wrapper
• 其余 6 后端：统一 JSON wrapper {"headers": {...}, "messageId": "uuid", "payload": "json-string"}

四、核心 API 草图

MessageSender<T>（15 方法）

覆盖 同步/异步/one-way × 普通/顺序 × 三种入参形式：

AbstractMessageSender<T>

• 持有 @Autowired BackendProducer backendProducer + ProduceConfig config
• 所有方法实现是一行薄代理：backendProducer.syncSend(config.toDestination(), msg)

MessageListener<T>

public abstract class MessageListener<T> implements InitializingBean {
    protected Type messageType;           // 反射从泛型 T 解析
    protected ConsumeConfig config;        // @Consumer 注解注入
    @Autowired private BackendConsumer backendConsumer;

    public abstract void onMessage(Headers headers, T message) throws Exception;

    @Override
    public void afterPropertiesSet() {
        backendConsumer.registerMessageListener(config, this);
        backendConsumer.start();
    }
}

BackendProducer SPI（12 + 1 方法）

Result syncSend(String destination, Message<?> msg);
Result syncSend(String destination, Object payload);
Result syncSendOrderly(String, Message<?>, String hashKey);
Result syncSendOrderly(String, Object, String hashKey);
void asyncSend(String, Message<?>, Callback);
void asyncSend(String, Object, Callback);
void asyncSendOrderly(String, Message<?>, Callback, String hashKey);
void asyncSendOrderly(String, Object, Callback, String hashKey);
void sendOneWay(String, Message<?>);
void sendOneWay(String, Object);
void sendOneWayOrderly(String, Message<?>, String);
void sendOneWayOrderly(String, Object, String);
void shutdown();

BackendConsumer SPI（4 方法）

void registerMessageListener(ConsumeConfig config, MessageListener<?> listener);
void start();
void stop();
void shutdown();

注解

@Producer(topic = "order-events", tag = "created")   // → AbstractMessageSender 子类
@Consumer(topic = "order-events", tag = "*", threadMin = 4, threadMax = 16)  // → MessageListener 子类

@Consumer 由 MessageListenerBeanPostProcessor（@Order(HIGHEST_PRECEDENCE)）在 postProcessBeforeInitialization 阶段注入 ConsumeConfig。

五、SPI 层与 7 后端实现

六、自动配置体系

flowchart TD
    subgraph "Backend AutoConfiguration（互斥 @ConditionalOnMissingBean）"
        B1["RocketMqMessagingAutoConfiguration<br/>@ConditionalOnProperty(easy.rocketmq.enabled)"]
        B2["RedisMessagingAutoConfiguration<br/>@ConditionalOnProperty(easy.redis.enabled)"]
        B3["KafkaMessagingAutoConfiguration<br/>@ConditionalOnProperty(easy.kafka.enabled)"]
        B4["RabbitMqMessagingAutoConfiguration"]
        B5["ZeroMqMessagingAutoConfiguration"]
        B6["DbmsMessagingAutoConfiguration"]
        B7["AutoMqMessagingAutoConfiguration"]
    end

    M["MessagingAutoConfiguration<br/>汇聚锚点<br/>注册 MessageListenerBeanPostProcessor"]

    B1 -->|"@AutoConfigureBefore"| M
    B2 -->|"@AutoConfigureBefore"| M
    B3 -->|"@AutoConfigureBefore"| M
    B4 -->|"@AutoConfigureBefore"| M
    B5 -->|"@AutoConfigureBefore"| M
    B6 -->|"@AutoConfigureBefore"| M
    B7 -->|"@AutoConfigureBefore"| M

    M -->|"@AutoConfigureAfter"| EVENT_BUS["CqrsMessagingAutoConfiguration<br/>easy-event-bus 适配层（类名为历史命名）"]

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

七、消息序列化与反序列化

Producer 侧

flowchart LR
    P1["AbstractMessageSender.send(T)"] --> P2["toMessageHeaders()"]
    P2 --> P3{"Connector 类型"}
    P3 -- "RocketMQ" --> P4["RocketMQ 原生 Message"]
    P3 -- "其余 6 后端" --> P5["JsonMapper.toJson(payload)<br/>+ 构造 JSON wrapper"]
    P5 --> P6["发送到 MQ"]

JSON wrapper 格式：{"headers": {...}, "messageId": "uuid", "payload": "json-string"}

Consumer 侧

flowchart LR
    C1["MessageListenerBeanPostProcessor<br/>注入 ConsumeConfig"] --> C2["MessageListener.afterPropertiesSet()"]
    C2 --> C3["backendConsumer.registerMessageListener()"]
    C3 --> C4["MQ 收到消息"]
    C4 --> C5{"Connector 类型"}
    C5 -- "RocketMQ" --> C6["原生 Message.getBody() → JsonMapper.fromJson()"]
    C5 -- "其余 6 后端" --> C7["先解 wrapper JSON<br/>→ Class.forName(messageType)<br/>→ JsonMapper.fromJson(payload)"]
    C6 --> C8["onMessage(headers, T)"]
    C7 --> C8

八、已知限制

九、关键设计决策