2. MQ 核心角色与代码对应

一、💡 一句话理解

核心结论

Producer 把业务数据包装成 Message 并发送给 Broker；Broker 按 Topic 分类、通过 Queue 保存和投递；Consumer 获取消息并执行业务逻辑。

Producer → Message → Broker → Topic 中的 Queue → Consumer

二、🧭 用“员工 1001 已离职”串起所有角色

沿用 1. MQ 基础中的例子：员工 1001 离职后，账户中心需要通知 ACL、HRMS 等下游系统。

{
  "employeeId": "1001",
  "status": "DEPARTED"
}

角色	在这个场景中是什么
Message	“员工 1001 已离职”这份账户数据
Producer	账户中心，也就是 `aboss-sso` 中发送消息的程序
Broker	SOFAMQ/RocketMQ 消息服务器
Topic	`ACCOUNT_CHANGE`，表示账户变更这一类消息
Queue	`ACCOUNT_CHANGE` 内实际保存消息的 Queue 0 或 Queue 1
Consumer	ACL 或 HRMS 中处理消息的程序实例

后面所有概念和代码都继续使用这个例子，不再切换到订单、支付等其他场景。

三、📨 Message：传递的那份数据

Message 通常包含：

Message
├── Topic：消息属于哪个业务分类
├── Tag：分类下进一步过滤
├── Key：可选的业务检索键，方便查询和排查
├── Body：真正的业务数据
└── 其他属性
    ├── 业务属性：生产者按需要主动添加
    └── 系统属性：消息 ID、发送时间等，由 MQ 生成

项目中创建消息的核心代码：

// 把账户详情转换成 JSON 字节数组，它就是消息体。
byte[] body = JSON.toJSONString(accountDetailDTO)
        .getBytes(StandardCharsets.UTF_8);
 
// 创建一条完整消息：Topic + Tag + Body。
Message message = new Message(topic, tag, body);

这段项目代码只设置了 Topic、Tag、Body，没有设置 Key 和自定义业务属性。它们是可选项，不设置也能正常发送消息。

如果希望使用员工 ID 查询消息，并额外标记事件类型，可以这样写：

Message message = new Message(topic, tag, body);
 
// Key 是业务检索键。出现问题时，可以按员工 ID 查询这条消息。
message.setKey(String.valueOf(accountDetailDTO.getAccId()));
 
// 自定义业务属性由生产者主动添加，消费者可以读取。
message.putUserProperties("eventType", "ACCOUNT_DEPARTED");

消息 ID、发送时间等系统属性通常由 MQ 客户端或 Broker 在发送过程中生成，所以不会出现在 new Message(...) 这几行代码里。

accountDetailDTO 是业务对象，Message 是 MQ 真正传输的对象。

放到当前例子里就是：

accountDetailDTO：员工 1001 已离职的账户数据
topic：ACCOUNT_CHANGE
key：1001，可用于查询和排查这名员工的消息
eventType：ACCOUNT_DEPARTED，自定义的事件类型
Message：MQ 真正发送的“员工 1001 已离职”消息

四、📤 Producer：负责发送消息

@Override
public void publish(AccountDetailDTO accountDetailDTO) {
    byte[] body = JSON.toJSONString(accountDetailDTO)
            .getBytes(StandardCharsets.UTF_8);
 
    Message message = new Message(topic, tag, body);
    getProducer().send(message);
}

getProducer() 不是 MQ 的固定语法，而是 AccountChangeMessage 重写的 Java 方法。当前项目中的真实实现是：

@Override
public Producer getProducer() {
    return producerSelector.select(name);
}

其中 name 来自下面的配置：

@Value("${spring.sofamq.normal.producers[1].name}")
private String name;

所以 getProducer() 的完整过程是：

读取 producers[1].name
  → 得到 accountChangeMessage
  → producerSelector.select("accountChangeMessage")
  → 返回已经创建好的 Producer

Producer 通常需要下面这些配置：

配置	解决什么问题	是否通常必需	当前项目对应配置
Producer 开关	决定项目启动时，是否创建用于发送普通消息的 Producer	根据项目框架决定	`spring.sofamq.normal.producers.enable=true`
MQ 接入地址	告诉 Producer 去哪里找 Broker	必需	`spring.sofamq.normal.endpoint=***`
实例或命名空间	指定消息属于哪套 MQ 环境	根据部署方式决定	`spring.sofamq.normal.instance-id=***`
AccessKey、SecretKey	证明当前应用有发送消息的权限	开启鉴权时必需	`spring.sofamq.normal.access-key=*`、`spring.sofamq.normal.secret-key=*`
Producer 名称	给 Producer 起一个项目内部名称，发送时用这个名称从 ProducerSelector 中取出 Producer 对象	当前项目框架需要	`spring.sofamq.normal.producers[1].name=accountChangeMessage`
Topic	指定消息属于哪个业务分类	发送消息时必需	`spring.sofamq.account.change.topic=TP_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY`
Tag	在 Topic 内进一步过滤消息	可选	`spring.sofamq.account.change.tag=TG_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY`
发送超时、重试次数	控制发送失败后如何处理	可选，但生产环境通常会关注	当前项目未发现显式配置，使用 SOFAMQ SDK 的默认策略

4.1 什么是普通消息

这里的“普通消息”不是说消息内容很普通，而是指它没有额外的顺序或事务要求：

消息类型	特点	发送方式示意
普通消息	发送后由 MQ 正常投递，不额外保证同一业务的处理顺序，也不和本地事务绑定	`Producer.send(message)`
顺序消息	同一业务键的消息需要按照发送顺序处理	使用顺序消息 Producer，并传入 ShardingKey
事务消息	消息是否提交，要和生产者的本地事务结果关联	使用事务消息 Producer 和事务回查

当前账户变更代码使用的是：

Producer producer = getProducer();
producer.send(message);

因此它属于普通消息。spring.sofamq.normal.producers.enable=true 的意思就是：项目启动时创建这种普通消息 Producer，后面的账户变更代码才能取得它并发送消息。

4.2 Producer 名称有什么用

Producer 名称 可以把它理解成项目内部保存 Producer 时使用的名字。ProducerSelector 的作用近似于一个 Map：

Map<String, Producer> producers = new HashMap<>();
 
// 项目启动时：用名称保存已经创建好的 Producer。
producers.put("accountChangeMessage", producer);
 
// 发送账户变更消息时：再根据名称把它取出来。
Producer producer = producers.get("accountChangeMessage");
producer.send(message);

当前项目的真实代码只是把 put/get 封装成了：

producerSelector.addProducer("accountChangeMessage", producer);
Producer producer = producerSelector.select("accountChangeMessage");

所以“标识并查找 Producer”更准确的意思是：**给 Java 对象起一个项目内部名称，以后按名称取出并复用它。**这个名称不是 Topic，也不是 Consumer Group。

4.3 当前项目怎样配置 Producer

在当前项目源码中，Producer 配置分散在公共配置和不同环境的配置文件中。

第一组是 Producer 总开关、Broker 连接和身份认证配置：

# 是否创建普通消息 Producer。
spring.sofamq.normal.producers.enable=true
 
# 访问 MQ 使用的身份信息，真实值不能写进学习笔记。
spring.sofamq.normal.access-key=***
spring.sofamq.normal.secret-key=***
 
# 要连接的 MQ 实例。
spring.sofamq.normal.instance-id=***
 
# MQ 接入地址。
spring.sofamq.normal.endpoint=***

第二组是两个 Producer 名称以及各自发送的消息分类：

# 钉钉消息 Producer。
spring.sofamq.normal.producers[0].name=dingDingMessage
spring.sofamq.dingding.topic=TP_CIC_MIDABOSS_BASIC_DINGTALK_TOPIC
spring.sofamq.dingding.tag=DINGTALK_WARN
 
# 账户变更消息 Producer。
spring.sofamq.normal.producers[1].name=accountChangeMessage
spring.sofamq.account.change.topic=TP_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY
spring.sofamq.account.change.tag=TG_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY

这些配置在 ProducerConfig 中真正参与了 Producer 的创建：

Properties credentials = new Properties();
credentials.setProperty(OMSBuiltinKeys.ACCESS_KEY, accessKey);
credentials.setProperty(OMSBuiltinKeys.SECRET_KEY, secretKey);
 
MessagingAccessPoint accessPoint = OMS.builder()
        .driver(MQ_DRIVER)
        .endpoint(endpointUrl)
        .withCredentials(credentials)
        .build();
 
Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.INSTANCE_ID, instanceId);
 
Producer producer = accessPoint.createProducer(properties);

创建完成后，项目把同一个 Producer 分别登记为 dingDingMessage 和 accountChangeMessage：

producerSelector.addProducer(dingDingName, producer);
producerSelector.addProducer(accountChangeName, producer);

发送账户变更消息时，getProducer() 再按 accountChangeMessage 这个名称把它取出来。Topic 和 Tag 则在创建每条 Message 时指定：

Message message = new Message(topic, tag, body);
getProducer().send(message);

可以理解为：Producer 的配置决定“快递员属于哪家公司、怎样连接仓库”，Message 中的 Topic 和 Tag 决定“这次包裹要送到哪个分类区”。

代码	含义
`accountDetailDTO`	原始业务数据
`new Message(...)`	把业务数据包装成消息
`getProducer()`	按 `accountChangeMessage` 这个名称取得已经创建好的 Producer
`send(message)`	把消息交给 Broker

Producer 通常不需要知道由哪台 Consumer 处理，也不需要知道有多少下游系统。

五、🏢 Broker：消息中转站

业务代码一般只写：

getProducer().send(message);

背后实际发生的是：

Producer.send(message)
    ↓
MQ 客户端找到 Broker
    ↓
Broker 接收 Message
    ↓
根据 Topic 找到消息分类
    ↓
选择 Topic 内的 Queue
    ↓
保存并等待 Consumer 消费

Broker 不只是简单的网络转发器。它还会保存消息和消费进度，并配合客户端完成消息分配；消费失败时，还要参与重新投递和死信处理。

这些内容分别放在后面的笔记中讲：

消费进度：不同 Consumer Group 分别记录自己消费到哪里。
负载均衡：同一个 Group 中的多个 Consumer 怎样共同分担 Queue。
重试和死信：消费失败后怎样重新投递，以及多次失败后怎样进入死信队列。

六、🗂️ Topic 和 Queue

6.1 Topic 是业务分类

ACCOUNT_CHANGE  账户变更消息
LOGIN_AUDIT     登录审计消息

员工 1001 离职属于账户变更，因此放进 ACCOUNT_CHANGE，不会放进 LOGIN_AUDIT。生产者向 Topic 发送消息，消费者订阅自己关心的 Topic。

6.2 Queue 是实际存储通道

在 RocketMQ 模型中，一个 Topic 可以包含多个 Queue：

ACCOUNT_CHANGE Topic
├── Queue 0：员工 1001 已离职、员工 1003 已离职
└── Queue 1：员工 1002 已离职、员工 1004 已离职

也就是说，Topic 先确定消息属于“账户变更”这一类，再由 Broker 把消息放进该 Topic 下的一条 Queue。多个 Queue 可以让多个 Consumer 并行处理消息。

记忆方法

Topic 回答“这是什么业务消息”：这是 ACCOUNT_CHANGE。Queue 回答“这条消息实际排在哪条通道里”：它被放进 Queue 0 或 Queue 1。业务代码通常指定 Topic，具体 Queue 多数由 MQ 路由策略选择。

七、📥 Consumer：真正处理消息

// 根据配置创建 Consumer。
Consumer consumer = accessPoint.createConsumer(properties);
 
// subscribe：订阅指定 Topic 和 Tag，并注册收到消息后的处理逻辑。
consumer.subscribe(topic, tag, new MessageListener() {
    @Override
    public Action consume(Message message, ConsumeContext context) {
        // 消息体是字节数组，先恢复成 JSON 字符串。
        String body = new String(
                message.getBody(),
                StandardCharsets.UTF_8
        );
 
        // 再把 JSON 转换成业务 DTO。
        AccountDetailDTO account =
                JSONUtils.parseObject(body, AccountDetailDTO.class);
 
        // 执行业务逻辑。
        accountService.saveOrUpdate(account);
 
        // 业务真正完成后，告诉 Broker 消费成功。
        return Action.CommitMessage;
    }
});
 
// 启动 Consumer，开始获取消息。
consumer.start();

Consumer 的核心动作只有五个：

订阅 → 接收 → 反序列化 → 执行业务 → 返回处理结果

放到当前例子中就是：ACL 或 HRMS 订阅 ACCOUNT_CHANGE，收到“员工 1001 已离职”后更新自己的权限或人员状态，处理完成后再返回成功结果。

八、✅ ACK 与消费结果

return Action.CommitMessage;

表示业务处理成功，可以推进当前消费组的消费进度。

return Action.ReconsumeLater;

表示本次没有处理成功，希望 Broker 稍后重新投递。

不能先确认再执行业务

如果数据库更新失败，却仍返回 CommitMessage，Broker 会认为消息已成功处理，通常不会再自动投递。

失败重试、死信、顺序、堆积和一致性继续看 5. MQ 可靠性问题；消费端如何防止重复修改数据继续看 6. 消息幂等。

九、📌 快速回顾

概念	一句话记忆
Message	要传递的数据
Producer	创建并发送消息
Broker	接收、保存、路由和投递
Topic	消息的业务分类
Queue	Topic 内实际存储和并行处理的通道
Consumer	获取消息并执行业务逻辑

下一篇：3. MQ Consumer Group 与消费模式。

Afyke的个人博客

探索

2. MQ 核心角色与代码对应

一、💡 一句话理解

二、🧭 用“员工 1001 已离职”串起所有角色

三、📨 Message：传递的那份数据

四、📤 Producer：负责发送消息

4.1 什么是普通消息

4.2 Producer 名称有什么用

4.3 当前项目怎样配置 Producer

五、🏢 Broker：消息中转站

六、🗂️ Topic 和 Queue

6.1 Topic 是业务分类

6.2 Queue 是实际存储通道

七、📥 Consumer：真正处理消息

八、✅ ACK 与消费结果

九、📌 快速回顾

十、🔗 官方资料

目录