一、💡 一句话理解
核心结论
本篇统一使用当前项目的 SOFAMQ + OpenMessaging Java SDK +
aboss-sso账户变更链路。可靠性不是打开一个开关，而是让生产端取得SendResult、SOFAMQ 保存并投递消息、消费者正确返回Action，再用幂等、重试、死信、对账和补偿共同兜底。
先记住这条主线：
防丢：SendResult + 本地消息表/事务消息 + CommitMessage/ReconsumeLater
防重：eventId + 消费记录唯一约束 + 业务幂等
保序：OrderProducer + 相同 accId 作为 ShardingKey
治堆积：查看 Group ID 消费状态 + 定位慢消费 + 扩容或提速
保一致：事务消息或本地消息表 + 对账补偿前置概念可先看 MQ 基础，当前项目的真实调用链见 SOFAMQ 账户变更代码实践，消费幂等的深入说明见 消息幂等。
二、🧭 理论：问题是什么
2.1 消息为什么会丢
先看当前项目里一条账户变更消息要走多远：
aboss-sso 修改账户
→ 把账户变更消息发送给 SOFAMQ
→ SOFAMQ 把消息交给 ACL 和 HRMS
→ ACL、HRMS 更新自己的账户数据这条路上有三个地方可能丢消息：
| 在哪里丢 | 举个例子 | 最后会怎样 |
|---|---|---|
| 发送时丢 | aboss-sso 发送失败，但代码只打印了一条错误日志 | aboss-sso 已经改了，ACL、HRMS 不知道 |
| SOFAMQ 中丢 | 消息长期没人消费，最后超过保存时间 | ACL、HRMS 永远收不到这条旧消息 |
| 消费时丢 | ACL 处理失败，却告诉 SOFAMQ“我处理成功了” | SOFAMQ 不再重发，但 ACL 实际没有更新成功 |
一句话理解
发送方要保证“发出去”，SOFAMQ 要保证“存得住”，消费方要保证“处理完”。三段中的任何一段没有兜住，都可能丢消息。
2.2 消息为什么会重复
重复消息通常不是 SOFAMQ 平白无故多造了一条，而是系统为了避免丢失进行了重试。
例如 ACL 收到一条账户变更消息：
1. ACL 已经把账户更新成功
2. ACL 正要告诉 SOFAMQ“处理完成”时，网络断了
3. SOFAMQ 没收到成功结果，以为 ACL 没处理
4. SOFAMQ 再投递一次
5. ACL 又收到同一条消息生产端也一样：SOFAMQ 其实已经收到了消息，但返回结果在路上丢了。aboss-sso 不知道到底成功没有，为了不丢，只能再发一次。
只要发送方或 SOFAMQ 没有收到明确的成功结果，就只能再执行一次。这样消息不容易丢，但消费者可能收到两次。
所以可靠消息通常选择：**宁可重复，也尽量不要丢失。**消费者再通过幂等保证重复消息不会重复产生业务效果。
2.3 消息为什么会乱序
假设同一个账户按下面的顺序发生变化：
账户启用 → 修改组织 → 账户停用正常情况下，下游也应该按这个顺序处理。但如果三条消息被不同线程同时消费，或者“修改组织”处理失败后重试，实际顺序可能变成：
账户启用 → 账户停用 → 修改组织这就是乱序。最后那条较旧的“修改组织”消息甚至可能把“账户停用”后的新数据覆盖掉。
MQ 所说的保序，通常不是让所有账户共用一条队伍，而是只保证同一个账户的消息有序。具体怎样用 SOFAMQ 顺序消息实现，放到 4.2 再讲。
2.4 消息为什么会堆积
当消息产生得比处理得快，没来得及处理的消息就会越来越多，这就是消息堆积。
可以把 SOFAMQ 想成一个水池：
aboss-sso 不断发送消息 = 进水
ACL、HRMS 不断处理消息 = 出水
SOFAMQ 保存等待处理的消息 = 水池短时间进水快没有问题；如果长时间进水快、出水慢，水池里的消息就会越来越多。
常见原因包括：
- ACL 或 HRMS 服务停了，没有人处理消息。
- 数据库慢，处理一条消息需要很久。
- 消费者数量太少，处理速度赶不上发送速度。
- 某条错误消息一直失败、一直重试。
堆积本身不一定是故障。真正需要警惕的是：堆积量持续增长，而且最老的消息等待时间也越来越长。
2.5 数据为什么会不一致
账户数据更新和发送消息是两个动作，只要其中一个成功、另一个失败，上下游数据就会不一致。
情况一：aboss-sso 账户修改成功 → 消息发送失败
结果：aboss-sso 是新数据，ACL、HRMS 还是旧数据
情况二：消息发送成功 → aboss-sso 账户修改回滚
结果：ACL、HRMS 根据消息改了数据，aboss-sso 自己却没改成功所以只调整顺序没有用：
- 先改数据库，再发消息，可能“数据库成功、消息失败”。
- 先发消息，再改数据库，可能“消息成功、数据库失败”。
这里追求的是最终一致：短时间不一致可以接受，但系统必须能通过重试、对账和补偿，最终让 aboss-sso、ACL、HRMS 的数据重新一致。
2.6 幂等是什么
幂等的意思是：同一条账户变更消息处理一次和处理多次，最终结果一样。
例如，一条消息要求把账户状态设置为“停用”：
处理 1 次：账户变成停用
处理 2 次：账户仍然是停用这比较容易做到幂等。但如果业务是“给失败次数再加 1”，重复处理两次就会多加一次，不是幂等。
判断消息是否处理过，不能只看 accId。因为同一个账户今天修改组织、明天修改手机号，这两次都是合法的新操作。
更稳妥的做法是：给每一次账户变更生成一个唯一的 eventId。
| 标识 | 代表什么 | 能不能用于判断同一次变更 |
|---|---|---|
accId | 哪个账户 | 不能，同一账户可以变更多次 |
eventId | 哪一次账户变更 | 可以，同一次补发继续使用同一个 eventId |
消费者处理前先检查 eventId：处理过就直接结束，没处理过才更新业务数据。这样即使 SOFAMQ 重复投递，业务也只会真正生效一次。
第二章快速记忆
丢失是消息没走完全程；重复是为了防丢而再次投递；乱序是先发生的后处理；堆积是产生快、处理慢；不一致是上下游结果不同；幂等负责让重复消息不重复生效。
三、⚙️ 理论：如何保证消息可靠性
可靠链路总览
总体方案先不要记一堆 API，只看一条消息怎样安全地走完全程：
aboss-sso 修改账户
→ 保证消息发得出去
→ SOFAMQ 保证消息存得住
→ ACL、HRMS 保证消息处理完成这条链路分成三层。
第一层：生产端保证“发得出去”
aboss-sso 发送账户变更消息后，不能只当作“发送方法已经调用过”，还要知道 SOFAMQ 是否已经收到。
- 发送成功：记录 SOFAMQ 返回的消息编号，方便以后查询。
- 发送失败：保存待发送记录，之后继续补发，不能只打印错误日志。
- 应用宕机：应用重启后仍然能够从待发送记录中找到这条消息。
这一层主要解决：**消息还没到 SOFAMQ 就丢了。**具体实现放在 3.1。
第二层：SOFAMQ 保证“存得住”
SOFAMQ 收到消息后，负责保存消息，并分别投递给 ACL Group 和 HRMS Group。
业务系统需要保留消息编号、关注消息保存时间和服务告警。这样消息没被消费时，还能查到它走到了哪里。
这一层主要解决：**生产者明明发送成功，但消息在中间没有被可靠保存或长期无人处理。**具体说明放在 3.2。
第三层：消费端保证“处理完”
ACL、HRMS 收到消息后，必须先完成数据库更新，再告诉 SOFAMQ“处理成功”。
- 业务成功：告诉 SOFAMQ 已经处理完成。
- 数据库临时故障：告诉 SOFAMQ 稍后再投一次。
- 消息本身有问题：保存错误信息并告警，避免一直无意义地重试。
这一层主要解决：**消息已经到达消费者，但业务没有真正处理成功。**具体实现放在 3.3。
消息走完整条链路后，还要分别处理四个附带问题：
- 重复：同一次账户变更用同一个
eventId，处理过就不再重复更新。 - 乱序：同一个
accId的消息排在同一条顺序队伍里，避免旧变化覆盖新变化。 - 堆积：监控还有多少消息没处理、最老的消息等了多久，再决定修复故障、提速或扩容。
- 数据不一致：用事务消息或本地消息表保证消息最终发出，再通过对账和补偿让
aboss-sso、ACL、HRMS 最终一致。
总体方案只记一句话
生产端发得出，SOFAMQ 存得住，消费端处理完；重复、乱序、堆积和不一致再分别治理。
3.1 生产端：发消息不能只按一下发送键
先看当前项目最危险的一种情况：
aboss-sso 已经把账户改成功
→ 准备发送账户变更消息
→ SOFAMQ 发送失败
→ 代码只打印一条错误日志
→ 方法继续结束
→ ACL、HRMS 永远不知道账户变了所以生产端真正要保证的不是“执行过 send()”，而是：这条消息最终一定有机会发送成功。
3.1.1 先看登录例子，再理解 send()
先看一个平时最常见的操作：登录系统。
输入账号密码，点击登录
→ 页面停下来等待服务器处理
→ 登录成功：服务器返回用户信息，页面进入系统
→ 登录失败或超时：页面提示错误在服务器返回结果之前，页面不会把这次登录当成成功。同步发送也是同样的过程。
当前项目执行 send(message) 时，程序会停在这一行等待 SOFAMQ 返回结果。这个过程叫同步发送。
结果只有两种：
SOFAMQ 接收成功
→ send() 返回一个 SendResult 对象
→ SendResult 里面有 messageId，可以用来查询这条消息
SOFAMQ 接收失败或长时间没有返回
→ send() 抛出异常
→ 代码进入 catch所以 SOFAMQ 不需要额外打开一个“Producer Confirm”开关。send() 本身就在等待结果。
项目里的 Producer 开关只表示“项目启动时创建一个普通消息 Producer”，不表示消息发送失败后会自动替业务补偿。
3.1.2 当前代码的问题：失败后只记日志
当前代码在发送失败后会捕获异常，但后续只剩下一行错误日志。问题不在于有没有捕获异常，而在于捕获以后没有留下可补发的数据：
- 没有保存哪条消息失败了。
- 没有安排下次什么时候重试。
- 应用一重启，这次失败只能靠人翻日志发现。
- 上层业务可能仍然认为整个账户修改已经完成。
日志只能告诉人“出过事”，不能保证消息“以后一定补得回来”。
3.1.3 可靠做法：先保存待发送记录，再发送
可以先把还没发送的消息保存到数据库，这张表叫本地消息表：
第一步：修改账户时，同时记下一条“账户变更消息待发送”
第二步：后台任务从本地消息表中取出消息，发送给 SOFAMQ
第三步：拿到 SendResult，把这条记录改成“发送成功”
第四步：发送失败就保留这条记录，过一会儿继续重试这里最关键的是第一步：账户数据和待发送记录要在同一个数据库事务里保存。
这样结果只有两种：
- 账户修改成功：数据库里一定有一条待发送记录，哪怕应用重启也能继续补发。
- 账户修改失败：待发送记录一起回滚，不会发出一条错误消息。
当前实现与建议方案
当前项目还没有完整的本地消息表，真实消息体仍是
AccountDetailDTO。这里讲的是可靠性改造方向，不是在描述当前已经具备的能力。
3.1.4 为什么这样做仍可能重复
假设 SOFAMQ 已经收到消息，但返回结果在路上丢了：
SOFAMQ 已收到消息
→ aboss-sso 没收到 SendResult
→ 以为发送失败
→ 再发一次
→ ACL、HRMS 可能收到两次为了避免丢失，生产端只能重发。因此每次账户变更都要有唯一 eventId，补发时继续使用原来的 eventId；ACL、HRMS 再根据它判断这次变更是否已经处理过。
3.1.5 eventId 是什么
先看一个账户连续修改两次的例子：
10:00，把账户 A100 的手机号改了
→ 这是一次账户变更，编号是 event-001
10:05，又把账户 A100 的部门改了
→ 这是另一次账户变更，编号是 event-002这里的 eventId，就是某一次业务变化的唯一编号。
accId回答的是：哪个账户变了？eventId回答的是：这是这个账户的哪一次变化？
如果 event-001 第一次发送超时，后台任务补发时仍然使用 event-001，不能重新生成 event-003：
ACL 第一次收到 event-001
→ 没处理过，修改手机号并记下 event-001
ACL 又收到 event-001
→ 已经处理过，不再修改手机号因此，eventId 不是 SOFAMQ 自动生成的字段，而是 aboss-sso 在发生账户变更时自己生成的业务编号。它应当同时保存到：
- 本地消息表，保证补发时还能找到原来的编号。
- 消息内容中，让 ACL、HRMS 能读取这个编号并去重。
- SOFAMQ 的 Message Key 中，方便按业务编号查询消息轨迹。
它与 messageId 也不是一回事：
| 名称 | 谁生成 | 表示什么 | 主要用途 |
|---|---|---|---|
eventId | aboss-sso | 某一次账户变更 | 业务去重、补发、对账 |
messageId | SOFAMQ 服务端 | SOFAMQ 中的一条消息 | 查询消息和排查链路 |
eventId 可以使用 UUID、雪花 ID，或者项目中已有且真正唯一的业务流水号。关键不在于采用哪一种格式，而在于满足两个条件：不同变更不能重复，同一次变更补发时不能改变。
当前项目与建议方案
当前消息体仍是
AccountDetailDTO。如果里面没有eventId，说明当前项目还没有完整实现这套业务幂等方案，需要在可靠性改造时新增，而不是认为 SOFAMQ 会自动提供它。
3.1.6 前端点了多次，怎样保证还是同一个 eventId
先看最容易出问题的情况：
用户点击“保存”
→ 后端其实已经修改成功
→ 返回结果时网络超时
→ 前端以为失败，用户又点了两次“保存”如果后端每收到一次请求就重新生成 eventId，结果会变成：
第一次请求 → event-001
第二次请求 → event-002
第三次请求 → event-003虽然三次请求修改的是同一份数据，但它们的 eventId 不同，后端和消费者就会把它们当成三次不同的业务操作。因此，只增加 eventId 还不够。
还要给本次前端提交增加一个 requestId，也叫幂等键。可以把它理解为“这次保存操作的编号”。
用户准备提交账户修改
→ 前端只生成一次 requestId：request-001
第一次提交
→ 携带 request-001
网络超时后再次提交
→ 仍然携带 request-001
用户连续点击多次
→ 仍然携带 request-001后端收到请求后，根据 requestId 判断：
第一次收到 request-001
→ 修改账户
→ 生成并保存 event-001
→ 保存 request-001 与 event-001 的对应关系
再次收到 request-001
→ 查到它已经处理过
→ 不再修改账户，也不再创建新消息
→ 直接返回第一次的处理结果和 event-001如果当前场景是“一次账户修改只产生一条 MQ 消息”，还可以直接让 requestId 和 eventId 使用同一个值，流程会更简单：
前端本次提交编号：request-001
后端本次事件编号：request-001
本地消息表补发编号：request-001
ACL、HRMS 去重编号：request-001前端的核心写法是：**同一次操作只生成一次编号，重试时不要重新生成。**下面是示意代码，不代表当前项目已经这样实现：
// 用户开始一次新的账户编辑时生成一次。
// 后续超时重试或重复点击，都继续使用这个值。
const requestId = crypto.randomUUID();
async function submitAccount(formData) {
await updateAccount({
requestId,
...formData
});
}这个编号要保留到后端明确返回成功为止。网络超时时，不能删除或更换它；只有用户开始下一次真正的新修改时，才生成新的 requestId。如果页面刷新后仍要支持继续重试，可以暂时把它保存在页面状态或 sessionStorage 中。
后端必须给 request_id 加唯一约束，不能只先查询再判断。因为两个重复请求可能在同一时刻到达，两边查询时都认为“还没处理过”。数据库唯一约束才能拦住并发重复：
-- 表名只是示意，实际可以放在本地消息表或业务请求记录表中。
ALTER TABLE mq_outbox
ADD UNIQUE KEY uk_request_id (request_id);完整保证由三层共同完成：
| 层次 | 做什么 | 能不能单独保证 |
|---|---|---|
| 前端按钮置灰 | 避免用户手快连点 | 不能，刷新、重试或并发请求仍可能发生 |
前端复用 requestId | 告诉后端这些请求属于同一次操作 | 不能，还需要后端校验 |
| 后端数据库唯一约束 | 并发请求也只能成功创建一条业务记录和一条消息 | 能提供最终保证 |
后端无法凭空判断两次请求是不是同一次操作
如果前端每次点击都生成新的
requestId，后端看到的就是两个不同请求。仅比较accId和请求内容也不可靠，因为用户可能确实连续做了两次相同修改。要识别超时重试，必须让调用方重复使用第一次的requestId，再由后端唯一约束兜底。相同requestId如果携带了不同的修改内容，后端应直接拒绝并告警，避免错误复用旧结果。
3.1.7 改造后的 MQ 发送代码
不用把所有新增类一次看完。这里只看 MQ 发送最关键的两步：
业务事务中：判断 requestId → 保存到本地消息表
后台任务中：读取本地消息 → 发送 SOFAMQ → 更新发送状态第一段：判断 requestId，并把消息保存到本地消息表
下面对应当前项目的 ExternalAccountUpdate4FrontCmdExe.execute()。账户具体怎样修改仍沿用原代码，这里只展开和 MQ 有关的部分：
@Transactional(rollbackFor = Exception.class)
public String execute(ExternalAccountUpdate4FrontCmd cmd) {
String requestId = cmd.getRequestId();
// 1. eventId 表示这一次账户变更。
// 当前场景一次请求只产生一条消息，所以直接让 eventId 等于 requestId。
String eventId = requestId;
// 2. 判断 requestId 是否已经处理过。
// tryCreatePreparing 会先向本地消息表插入一条 PREPARING 记录，
// request_id 有唯一索引，所以并发请求也只能有一个插入成功。
boolean firstRequest = mqOutboxService.tryCreatePreparing(
requestId,
eventId
);
if (!firstRequest) {
// 插入失败，说明同一个 requestId 已经来过。
// 不再修改账户，也不再创建第二条 MQ 消息。
MqOutboxRecord oldRecord =
mqOutboxService.findByRequestId(requestId);
return oldRecord.getAccId();
}
// 3. 沿用当前项目原来的账户修改逻辑。
Account account = init(cmd);
update(account);
modify(account);
// 4. 查询修改后的完整账户数据，作为 MQ 消息内容。
AccountDetailDTO accountDetail =
accountByAccountIdQryExe.execute(
new AccountIdQry()
.setAccId(account.getAccountId())
);
AccountChangedEvent event = new AccountChangedEvent()
.setEventId(eventId)
.setAccount(accountDetail)
.setOccurredAt(new Date());
// 5. 把刚才的 PREPARING 记录补充完整，并改成 PENDING。
// 此时还没有调用 SOFAMQ，只是保存了一条“等待发送”的消息。
// 这一步和账户修改处于同一个数据库事务：
// 账户修改失败，两者一起回滚；
// 账户修改成功，待发送消息一定会保留下来。
mqOutboxService.markPending(
requestId,
eventId,
account.getAccountId(),
JSON.toJSONString(event)
);
return account.getAccountId();
}上面仍然需要给本地消息表的 request_id 加唯一索引。真正拦住并发重复的是唯一索引，Java 判断只负责决定重复请求应该返回什么结果。
第二段：后台任务真正发送 SOFAMQ
后台任务不断查询状态为 PENDING 或 RETRY 的记录：
public void sendPendingMessages() {
List<MqOutboxRecord> records =
mqOutboxService.findPendingMessages(100);
for (MqOutboxRecord record : records) {
try {
// 1. 从本地消息表读取原来的消息体。
// messageBody 中已经包含 eventId，补发时不能重新生成。
Message message = new Message(
topic,
tag,
record.getMessageBody()
.getBytes(StandardCharsets.UTF_8)
);
// 2. 同步发送 SOFAMQ。
// 程序会停在这里，等待 SOFAMQ 返回 SendResult。
SendResult sendResult = getProducer().send(message);
// 3. SOFAMQ 明确返回成功。
// 把本地消息改成 SENT，并保存 messageId 方便查询轨迹。
mqOutboxService.markSent(
record.getId(),
sendResult.getMessageId()
);
} catch (Exception e) {
// 4. 发送失败或超时。
// 不删除本地消息，也不生成新的 eventId。
// 只把状态改成 RETRY，等待后台任务下次继续发送。
mqOutboxService.markRetry(
record.getId(),
e.getMessage()
);
log.error(
"账户变更消息发送失败，eventId={}，等待补发",
record.getEventId(),
e
);
}
}
}把两段代码连起来看：
第一次提交 request-001
→ eventId 使用 request-001
→ 消息保存到本地消息表，状态为 PENDING
发送 SOFAMQ 超时
→ 本地记录改成 RETRY
→ eventId 仍然是 request-001
后台再次补发
→ 读取同一条本地记录
→ 发送的仍然是 request-001当前代码与改造代码
当前项目还没有 MqOutboxService 和本地消息表。上面展示的是在现有 ExternalAccountUpdate4FrontCmdExe、AccountChangeMessage 基础上应该改成的核心发送逻辑，只保留了理解可靠发送必需的代码。
3.1 只记这一句
requestId和唯一约束负责拦住前端重复提交；SendResult负责告诉你这次有没有拿到发送结果；本地消息表负责保证失败以后还能继续补发；eventId负责处理消息重复。
3.2 SOFAMQ 服务端保证消息存得住
3.2.1 先用上传网盘理解
假设你把一个文件上传到网盘：
点击上传
→ 网盘服务器接收文件
→ 网盘明确返回“上传成功”
→ 文件保存在网盘服务器
→ 以后可以在另一台电脑下载这里有两个不同的成功：
- 上传成功：网盘服务器已经收到了文件。
- 下载成功：另一台电脑已经把文件下载并使用了。
SOFAMQ 也是一样：
aboss-sso 调用 send(message)
→ SOFAMQ 接收并保存消息
→ send() 返回 SendResult
→ SOFAMQ 再把消息投递给 ACL、HRMS因此，拿到 SendResult 只能说明：**SOFAMQ 服务端已经接收了消息。**它不能说明 ACL、HRMS 已经修改完数据库。
3.2.2 SOFAMQ 收到消息后做了什么
SOFAMQ 服务端主要包含 NameServer 和 Broker：
- NameServer：保存 Topic 和 Broker 地址的对应关系，可以理解成“地址导航”。
- Broker：真正接收、保存并转发消息，可以理解成“消息仓库”。
先用去商场买相机理解：
你先看商场导购牌
→ 导购牌告诉你：相机柜台在 3 楼 A 区
→ 你再去 3 楼 A 区购买相机这里：
| 商场例子 | SOFAMQ |
|---|---|
| “相机”这个商品分类 | Topic |
| 商场导购牌 | NameServer |
| 3 楼 A 区相机柜台 | Broker |
导购牌只告诉你柜台在哪里，它自己不卖相机；同样，NameServer 只告诉程序 Broker 的网络地址，它自己不接收、不保存消息。
假设账户变更 Topic 的数据由 Broker A 和 Broker B 负责保存，NameServer 中记录的关系大致是：
账户变更 Topic
→ Broker A：10.0.0.1:端口
→ Broker B：10.0.0.2:端口aboss-sso 启动 Producer 后，会先向 NameServer 查询这个 Topic 的地址，然后直接连接对应的 Broker 发送消息：
aboss-sso 问 NameServer：
“账户变更 Topic 在哪里？”
NameServer 返回：
“去 Broker A 或 Broker B。”
aboss-sso 再把消息发送给 Broker
→ Broker 接收消息
→ Broker 保存消息
→ Broker 后续投递给 ACL、HRMS所以，“某个 Topic 应该去找哪些 Broker”的完整意思是：
这个 Topic 的消息由哪些 Broker 服务器负责保存？这些服务器的 IP 和端口是什么？
Topic 只是消息分类名称，不是一台服务器；NameServer 负责提供地址，Broker 才负责保存消息。
服务端采用集群方式部署。Broker 有 Master、Slave，生产者和消费者也可以部署多个实例。某台服务器或某个机房发生故障时，平台通过集群继续提供服务。
这些能力由 SOFAMQ 平台负责，业务代码不用再配置 RabbitMQ 的持久化队列、Kafka 的副本数等其他产品参数。
但“平台负责保存”不等于“业务什么都不用管”。业务项目仍要保证：
消息发送到了正确的 SOFAMQ 实例
→ 进入正确的 Topic 和 Tag
→ 正确的 Group ID 正在消费
→ 消费异常能够及时发现3.2.3 Topic、Tag、Group ID、Instance 和 Endpoint 是什么
可以把它们理解成消息的地址：
| 配置 | 大白话作用 | 配错以后会怎样 |
|---|---|---|
Endpoint | 连接哪一个 SOFAMQ 接入地址 | 连接失败，或者连到错误环境 |
Instance ID | 使用哪个 SOFAMQ 实例 | 找不到当前环境的 Topic、Group，或发到错误实例 |
Topic | 消息属于哪个大分类 | 生产者和消费者 Topic 不一致，ACL、HRMS 收不到消息 |
Tag | Topic 下的进一步分类 | 消费者过滤掉这条消息，看起来像“消息丢了” |
Group ID | 哪一组消费者共同维护消费进度 | 配到错误 Group，消费进度、重试和堆积都会算到别的组 |
当前账户变更链路可以记成：
Topic：账户变更消息的大分类
Tag：账户修改这一种具体消息
ACL Group：ACL 自己的消费进度
HRMS Group：HRMS 自己的消费进度ACL Group 和 HRMS Group 不同，所以同一条账户变更消息会分别交给两个系统处理。Group ID 不是生产者发送时指定的“收件人”，而是消费者订阅 Topic 时使用的身份。
SOFAMQ 官方还要求 Group ID、Topic、Endpoint 和 Instance ID 使用同一地域、同一套环境的资源。测试环境和生产环境不能混用。
3.2.4 为什么要记录 messageId 和 Message Key
假设有人反馈：“账户已经修改，但 ACL 没有同步。”如果日志里只有“发送成功”，仍然不知道具体是哪一条消息。
SOFAMQ 提供三种查询方式：
| 查询条件 | 谁生成 | 适合什么时候使用 |
|---|---|---|
messageId | SOFAMQ 服务端 | 已经拿到发送结果，需要精确查询一条消息 |
| Message Key | 生产者设置 | 已知业务编号，例如 eventId，但不知道 messageId |
| Topic + 时间范围 | 人工选择范围 | 前两项都没有，只能在一段时间内查找 |
当前场景可以把 eventId 设置成 Message Key：
Message message = new Message(
topic,
tag,
messageBody.getBytes(StandardCharsets.UTF_8)
);
// eventId 是业务编号，用它可以在控制台按业务查询消息。
message.setKey(eventId);
// messageId 是 SOFAMQ 在发送成功后返回的消息编号。
SendResult sendResult = getProducer().send(message);
log.info(
"账户变更消息发送成功，eventId={}，messageId={}",
eventId,
sendResult.getMessageId()
);两者不要混淆：
eventId / Message Key：业务自己设置，用来认出“哪一次账户变更”
messageId：SOFAMQ 生成，用来认出“服务端里的哪一条消息”查询到消息后，还可以继续查看消息轨迹，确认它是否经过了下面这些节点：
生产者发送
→ SOFAMQ 服务端接收
→ 投递给 ACL Group
→ 投递给 HRMS Group
→ 各 Group 是否消费成功3.2.6 为什么 Broker 不能当成永久存储
SOFAMQ 官方使用限制中，消息保存时间是 3 天。超过保存时间后，旧消息会被自动滚动删除。
例如 ACL 故障了四天：
第一天：SOFAMQ 中还有待消费消息
第二天：消息仍然可以继续消费
第三天：已经接近保存上限
第四天：最早的消息可能已经被清理所以不能认为“消息已经发送到 Broker，以后任何时间都能重新消费”。
需要长期保留的是业务数据库、本地消息表、消费记录和对账数据。SOFAMQ 负责在有限时间内传递消息，不负责永久保存业务历史。
这也是为什么最终一致不能只靠 Broker，还需要：
- 监控长时间未消费的消息，不能等到超过保存时间才发现。
- 定期对账
aboss-sso、ACL、HRMS 的最终数据。 - 已经过期或丢失的历史消息，通过业务数据重新生成补偿消息。
保存时间以环境为准
官方使用限制页面标明为 3 天；实际排查时仍应在当前 SOFAMQ 环境的控制台或平台说明中再次确认。
3.2.7 应用到底要监控什么
可以把监控分成三段：
| 位置 | 重点监控 | 出现异常说明什么 |
|---|---|---|
aboss-sso 生产端 | 发送成功率、异常数、本地消息表 PENDING/RETRY/FAILED 数量 | 消息可能还没成功进入 SOFAMQ |
| SOFAMQ 服务端 | 服务告警、消息查询和消息轨迹 | 服务端接收、保存或投递链路可能异常 |
| ACL、HRMS Group | 是否在线、消费速度、堆积量、最近消费时间、消息延迟 | 消息已经到达服务端，但消费者处理不及时 |
其中最值得警惕的不是“当前有 100 条堆积”，而是下面这种趋势：
堆积量持续增加
+ 最老消息等待时间持续增加
+ 消费速度长期低于生产速度这通常说明消费者离线、数据库变慢、某条错误消息反复重试，或者消费能力确实不足。
3.2.8 发送成功但下游没更新，怎样排查
按照消息走过的顺序查，不要一上来就翻所有系统日志：
- 在
aboss-sso日志或本地消息表找到eventId和messageId。 - 在 SOFAMQ 控制台按
messageId精确查询；没有时再按 Message Key 查询。 - 查看消息轨迹，确认 SOFAMQ 是否收到、是否投递给 ACL Group 和 HRMS Group。
- 查看目标 Group 是否在线，以及堆积量、最近消费时间和消息延迟。
- 如果轨迹显示已经投递，再检查 ACL、HRMS 消费日志和业务数据库。
- 如果消息已经超过保存时间，只能通过业务对账确定差异并重新补偿。
3.2 只记这一句
SOFAMQ 负责在有限时间内把消息存住并投递；项目负责配置正确地址、留下查询编号、监控消费状态，并在消息过期前处理异常。
3.3 消费端保证业务处理完成
SOFAMQ 消费监听器通过返回 Action 告诉客户端 SDK 本次消费结果：
public Action consume(Message message, ConsumeContext context) {
try {
AccountChangedEvent event = deserialize(message);
// 消费记录与 ACL/HRMS 的业务更新必须处于同一数据库事务。
accountChangeService.consumeIdempotently(event);
// 业务事务成功：确认完成，不再向当前 Group 投递。
return Action.CommitMessage;
} catch (RetryableException e) {
// 数据库暂时不可用、下游超时等可恢复故障：稍后重试。
return Action.ReconsumeLater;
} catch (IllegalArgumentException e) {
// 永久性脏数据：保存原消息并告警，避免无意义重试。
badMessageService.save(message, e);
return Action.CommitMessage;
}
}| 返回结果 | SOFAMQ 后续动作 |
|---|---|
Action.CommitMessage | 当前 Group 消费成功，不再重投 |
Action.ReconsumeLater | 当前 Group 稍后重新消费 |
返回 null 或抛出异常 | 对普通集群消息同样会触发重试，但不如显式返回清楚 |
SOFAMQ 的普通集群消息默认最多重试 16 次，达到上限仍失败会进入该 Group ID 对应的死信队列。广播消费不提供失败重试能力。
当前项目的消费风险
如果 ACL、HRMS 捕获所有异常后仍返回
CommitMessage，SOFAMQ 会认为消息已经成功处理，不会自动重试。正确做法是区分临时故障和永久性脏数据。
四、🛠️ 理论：重复、乱序、堆积与最终一致怎么处理
4.1 如何解决消息重复
-
**刚开始，发送失败只记录日志，有些消息根本没到 MQ 服务器。**账户已经改成功了，但消息没发出去，下游也就不知道账户变了。
-
**所以我们在发送前，先把消息存进数据库，标记为“待发送”。**等 MQ 服务器返回成功后，再把它改成“发送成功”。同时加一个定时任务，不断扫描未发送成功的消息，继续补发。修改账户和保存待发送消息要放在同一个事务里，保证账户修改成功时，这条消息也一定存下来了。
-
**但有时候，消息已经到了 MQ 服务器，但是因为网络问题没返回给生产者成功。**生产者没收到成功回复，就不会把消息改成“发送成功”。定时任务下次扫描到它，又往 MQ 发了一次，这样就产生了重复消息，消费者可能会处理两次。
-
**所以我们给每一次业务变化设置一个唯一编号
eventId。**同一条消息补发时，继续用原来的eventId。消费者收到后，判断这个编号是不是已经处理过：没处理过才改数据；处理过就跳过业务更新。但不是直接扔掉不管，而是告诉 MQ：“这条我之前处理过了，算成功，别再发了。”判断是否处理过，要靠消费记录的唯一约束拦住并发重复，消费记录和业务更新也要在同一个事务里完成。 -
**后来又加了手动补偿，因为有些消息很着急，等不了定时任务扫描。**我们会手动点击补偿按钮，但网络抖动可能让请求重发，人也可能连续点好几次。按钮又不能靠点一次就置灰来拦，所以给这次补偿操作加一个
requestId，同一次操作的连续点击和重试都用这个编号。请求到后端时，第一次就保存请求并安排补偿；后面再收到相同编号，就返回原来任务的状态，不再重复安排。这里也要给requestId加唯一约束，不能只先查一下有没有。手动补发的消息仍然使用原来的eventId，这样即使和定时任务撞到一起，消费者也能认出是同一件事。如果每次点击都换一个requestId，它就拦不住。 -
**最后，如果有人一直发起新操作，带来的请求编号、事件编号都不同，就还得让数据库守住业务规矩。**比如业务规定手机号不能重复注册，就给手机号加唯一约束；如果只要求同一租户内不能重复，就约束租户和手机号的组合。只保证主键不重复没用，因为两个账户可以主键不同、手机号相同。这一层负责防止不同操作建出重复的业务数据，不能代替前面的
eventId去重。
4.2 如何保证同一账户的消息顺序
需要顺序时，不能继续使用普通 Producer，而要使用 SOFAMQ OrderProducer：
import io.openmessaging.api.SendResult;
import io.openmessaging.api.order.OrderProducer;
String shardingKey = accountDetailDTO.getAccId();
SendResult sendResult = orderProducer.send(message, shardingKey);完整思路是：
同一个 accId
→ 作为相同 ShardingKey
→ 进入同一个顺序分区
→ SOFAMQ 在分区内按顺序投递
→ 消费者再用版本号防止旧状态覆盖新状态顺序消息消费失败时会持续重试并阻塞同一分区，因此必须监控失败消息，不能让一条坏消息长期卡住后面的账户变化。
说白了就是：前面一条没处理成功，后面的先别动。
比如同一个账户，有三条消息，后面的操作都依赖账户已经创建：
① 创建账户
② 修改手机号
③ 停用账户消费者处理第 ① 条时，数据库报错了，账户没创建成功。如果这时候继续处理第 ② 条，就会发现：账户都没有，改谁的手机号？
所以顺序消费会这样做：
① 创建账户失败 → 重试① → 再失败 → 再重试①
② 修改手机号：等着
③ 停用账户：等着这就是“失败重试并阻塞”：**① 一直没成功，② 和 ③ 就一直等。**而且排队是按分区来的，假如另一个账户的消息也被分到这个分区，它也可能跟着等。
“不能让一条坏消息长期卡住”，意思是要及时发现并处理：
- 数据库暂时故障：修好后，① 重试成功，后面的继续。
- 消息里的数据本身就错了：比如创建账户缺少必填字段，重试多少次都不会成功，就需要人工排查、修正后再处理。
但不能直接把 ① 跳过去就算了，因为后面的 ② 和 ③ 可能依赖 ①。要先弄清楚跳过会不会出问题，再决定怎么处理。
那人工排查期间，就让后面的消息一直等着吗？
按上面这套直接重试的方案，是的，同一分区后面的消息就得等。这就是严格顺序消费的代价：前面的没成功，后面不能先做。
但如果账户 A 创建失败，排查要花几个小时，业务又不能接受其他账户跟着等，可以另外设计一套处理办法：
- **发现 A 的消息一直失败，把 A 标记成“暂停处理”。**这个状态要保存到数据库，让处理这个账户的消费者都能知道，不能只记在某台机器的内存里。
- **把 A 这条失败消息完整存进数据库。**之后收到 A 的修改、停用等消息，也按顺序存下来，先不执行业务。消息和暂停状态都可靠保存后，再告诉 MQ 这次投递已接收，让队列继续往下走；如果保存失败，就不能确认，仍要等待重试。此时只是把消息接过来保管，不能把它记成“业务已处理成功”，否则恢复时会被去重逻辑跳过。
- **其他账户照常处理。**只有 A 的消息留在数据库里等。不能只拿走 A 的“创建账户”，就放行 A 的“修改手机号”；同一个账户后面依赖它的消息，也必须一起等。
- **问题修好后，按原来的顺序处理 A 暂存的消息。**恢复期间，新收到的 A 消息也继续排到后面，由同一套按账户排队的逻辑处理，避免新消息抢到旧消息前面。确认暂存消息全部处理完后，再恢复正常处理；最后一条暂存消息与新消息交接时，也要保证没有两边同时执行。
这相当于：**A 出了问题，就让 A 自己排队等，别把其他账户一起堵住。**不过，这套暂停、存消息、按顺序恢复的逻辑，需要我们自己实现，MQ 不会自动替我们做。暂存消息恢复时仍要使用原来的 eventId，并在业务实际成功后，才把消费记录记为已完成。
这套办法适合“某个账户的消息有问题”。如果是大家共用的数据库都不可用了，连暂存消息也存不进去，就不能靠它放行，还是得先恢复数据库。
如果不增加这套逻辑，那就只能保留阻塞，及时告警、尽快修复。严格按顺序执行，总得有个地方替这个账户排队：要么等在 MQ 里，要么等在我们自己的数据库里。
当前项目还不是顺序消息
当前代码使用普通
Producer.send(message)。只有确认账户变更确实要求严格顺序，并完成 OrderProducer、Topic 和消费方式的配套改造后，才能认为顺序得到保证。
4.3 如何处理消息堆积
先在 SOFAMQ 控制台进入目标 Group ID 的消费者状态，查看：
- Consumer 是否在线。
- 实时消费速度。
- 实时消息堆积量。
- 最近消费时间。
- 消息延迟时间。
然后按顺序处理：
- 检查 ACL、HRMS 消费日志和
sofamq.log，确认是否存在持续异常。 - 检查数据库慢 SQL、锁等待、连接池和下游接口耗时。
- 隔离反复失败的消息，达到重试上限后从死信队列排查和重发。
- 在分区和消费能力允许的情况下增加 Consumer 实例或消费线程。
- 长期优化单条处理耗时、批量处理方式和数据库写入效率。
只增加实例不一定有效。如果并发上限受 Topic 分区数或数据库瓶颈限制，继续加实例也不会提高消费速度。
4.4 如何保证最终一致
先不要看 TransactionProducer 这些类名，先看这个问题到底在解决什么。
4.4.1 什么叫“最终一致”
假设用户在 aboss-sso 中停用一个账户，需要发生两件事：
第一件事：aboss-sso 把账户改为“停用”
第二件事：发送 MQ 消息，通知 ACL 和 HRMS 也停用这个账户这两件事不在同一个系统里，无法直接放进同一个普通数据库事务，因此可能出现下面的情况：
aboss-sso 修改成功
→ 正准备发送消息时，应用宕机
→ ACL、HRMS 没收到通知
→ 三个系统的数据不一致“最终一致”不是要求三个系统在同一毫秒完成修改，而是允许它们短时间不一致，但必须有重试、回查和补偿手段，让它们最后变成一致。
刚提交时：aboss-sso 已停用，ACL、HRMS 还没更新
稍后重试：消息发送成功，ACL、HRMS 完成更新
最终结果：三个系统都是停用在 SOFAMQ 技术栈下，生产端通常有两种实现办法。二选一即可，不是两种都必须使用。
4.4.2 方案一：本地消息表
可以把本地消息表理解为 aboss-sso 自己保存的一张“待通知清单”。
用户停用账户时，aboss-sso 在同一个数据库事务中完成两次写入：
1. 把账户改为“停用”
2. 在本地消息表写入一条“需要通知 ACL、HRMS”的记录因为两次写入属于同一个本地事务，所以结果只能是：
两条数据一起写入成功
或者
两条数据一起回滚只要账户真的修改成功，“待通知清单”中就一定会留下记录。后台任务不断扫描这张表，并使用普通 Producer.send(message) 发送消息，直到发送成功。
账户修改成功
→ 本地消息表留下 PENDING 记录
→ 第一次发送失败
→ 后台任务以后继续发送
→ SOFAMQ 接收成功
→ 本地消息记录改为 SENT这就是本地消息表能够保证最终一致的原因：即使应用在发送前宕机，数据库中仍然保存着“这条消息还没有发”的证据，恢复后可以继续补发。
不过，可能出现“SOFAMQ 已经收到消息，但本地记录还没来得及改成 SENT”的情况。后台任务会再次发送同一个 eventId，因此 ACL、HRMS 仍然必须根据 eventId 做幂等判断。
4.4.3 方案二：SOFAMQ 事务消息
先看普通消息为什么会出问题。
用户在 aboss-sso 中停用账户，程序通常会这样执行：
1. 数据库把账户改成“停用”
2. 调用 Producer.send(message) 通知 ACL、HRMS如果程序刚执行完第 1 步就宕机，第 2 步还没执行，结果就是：账户已经停用，但消息永远没有发出去。
SOFAMQ 事务消息就是专门堵住这道缝。可以先把它理解成一条关闭了投递开关的消息。
4.4.3.1 正常情况下怎么执行
还是用“停用账户”这个例子：
第一步：aboss-sso 先把“停用账户”消息交给 SOFAMQ
SOFAMQ 保存消息，但暂时关闭投递开关
所以 ACL、HRMS 现在还看不到它
第二步：确认 SOFAMQ 已经保存消息后
aboss-sso 才执行本地数据库事务，把账户改成“停用”
第三步：数据库修改成功
aboss-sso 返回 CommitTransaction
SOFAMQ 打开投递开关，把消息发给 ACL、HRMSSOFAMQ 官方把这种“已经保存，但暂时不能投递”的消息称为半事务消息（Half Message）。
这里最重要的不是“半”字，而是记住：
半事务消息
消息已经在 SOFAMQ 里了，只是暂时不让 ACL、HRMS 消费。
4.4.3.2 如果数据库修改失败呢
假设 SOFAMQ 已经保存了消息，但 aboss-sso 修改账户失败：
SOFAMQ 已经保存消息，投递开关仍然关闭
→ aboss-sso 数据库修改失败并回滚
→ 返回 RollbackTransaction
→ SOFAMQ 删除这条消息
→ ACL、HRMS 不会收到错误通知所以三个返回值可以直接理解成对 SOFAMQ 下命令：
| 返回值 | 对 SOFAMQ 说的话 | 最终处理 |
|---|---|---|
CommitTransaction | 账户修改成功，可以通知下游了 | 打开投递开关 |
RollbackTransaction | 账户修改失败，不要通知下游 | 删除消息 |
Unknow | 我现在不能确定，先别投递 | 保持关闭，稍后再问 |
4.4.3.3 为什么还需要回查
最容易出问题的是下面这个瞬间：
SOFAMQ 已经保存半事务消息
→ aboss-sso 数据库已经修改成功
→ 正准备返回 CommitTransaction 时，应用突然宕机这时 SOFAMQ 手里有消息，但没有收到最后答案。它不能直接投递，也不能直接删除，所以过一段时间会主动询问 aboss-sso：
SOFAMQ：eventId=E1001 这次账户修改，到底成功了吗？接住这个询问的代码，就叫 LocalTransactionChecker。它要根据数据库中保存的业务结果或事务记录回答：
查到 E1001 已经执行成功
→ 返回 CommitTransaction
→ SOFAMQ 投递消息
查到 E1001 已经执行失败
→ 返回 RollbackTransaction
→ SOFAMQ 删除消息
暂时仍然查不清
→ 返回 Unknow
→ SOFAMQ 以后继续回查因此，回查不是重新执行一次“停用账户”，而是查询上一次数据库事务到底成功还是失败。否则重复执行本地业务，反而可能造成数据错误。
4.4.3.4 对应到程序中的三个角色
| 程序对象 | 在这个例子里负责什么 |
|---|---|
TransactionProducer | 先发送半事务消息，再协调本地事务结果 |
LocalTransactionExecuter | 执行 aboss-sso 的账户修改，并返回成功、失败或未知 |
LocalTransactionChecker | 应用宕机或结果丢失后，查询原来的本地事务结果并回答 SOFAMQ |
整套流程可以压缩成一句话：
消息先存入 SOFAMQ，但先不投递
→ 再修改本地数据库
→ 修改成功就投递，修改失败就删除
→ 没收到结果，SOFAMQ 就回来查询它只解决生产端的问题
SOFAMQ 事务消息保证的是：
aboss-sso账户修改成功后，对应通知不会因为应用宕机而悄悄消失。它不代表 ACL、HRMS 已经修改成功；下游收到消息后的失败重试和eventId幂等仍然不能少。
4.4.4 当前项目应该怎么选
| 对比项 | 本地消息表 | SOFAMQ 事务消息 |
|---|---|---|
| 谁记住消息还没发 | aboss-sso 的本地消息表 | SOFAMQ 的半事务消息 |
| 失败后怎么恢复 | 后台任务扫描并补发 | SOFAMQ 调用 LocalTransactionChecker 回查 |
| 项目需要增加什么 | 消息表、补发任务、状态和告警 | TransactionProducer、本地事务执行器、回查逻辑和专用 Group ID |
| 排查是否直观 | 可以直接查本地消息表 | 需要结合业务记录和 SOFAMQ 消息轨迹排查 |
当前项目已经使用普通 Producer，并且前面已经设计了 eventId、本地消息表和补发任务，所以更容易落地的是本地消息表方案。
如果平台已经开通并统一支持 SOFAMQ 事务消息，也可以改用 TransactionProducer。但不能只把普通 Producer 换一个类名，还必须实现可靠的本地事务和 LocalTransactionChecker 回查逻辑。
4.4.5 为什么生产端做完还不够
本地消息表或事务消息只能尽量保证消息从 aboss-sso 发出去。消息到达 ACL、HRMS 后，还可能遇到数据库超时、应用重启或重复投递，因此消费端还要继续保护：
| 消费端处理 | 实际作用 |
|---|---|
业务处理成功后返回 CommitMessage | 告诉 SOFAMQ：当前 Group 已经处理完成 |
临时失败返回 ReconsumeLater | 告诉 SOFAMQ：这次没处理完，稍后再投递 |
根据 eventId 做幂等 | 同一条消息收到多次，也只执行一次业务修改 |
| 多次重试仍失败后进入死信队列 | 把一直失败的消息隔离出来，等待排查和人工补偿 |
| 定期对账三个系统 | 发现所有自动重试都没修复的漏网问题 |
把整条链路连起来，就是：
aboss-sso 修改账户
→ 通过“本地消息表”或“SOFAMQ 事务消息”保证通知最终能发出
→ ACL、HRMS 处理失败时让 SOFAMQ 重试
→ 重复投递时使用 eventId 防止重复修改
→ 自动重试仍未解决时进入死信队列并告警
→ 最后通过定期对账发现和补偿剩余差异4.4 只记这一句
生产端保证消息最终能够发出去，消费端保证消息最终能够处理成功，对账负责发现最后的漏网问题；这三层加起来，才叫最终一致。
五、🚀 实践：改造当前账户变更链路
5.1 为账户变更事件增加 eventId
建议不要直接把裸 AccountDetailDTO 当成完整事件，而是增加事件外壳：
{
"eventId": "account-change-20261016-000001",
"occurredAt": "2026-10-16T10:30:00+08:00",
"account": {
"accId": "A10001",
"isValid": true
}
}eventId 表示一次具体的账户变更；accId 表示哪个账户。两者职责不同。
如果一次账户提交只产生一条 MQ 消息，可以直接使用前端本次提交的 requestId 作为 eventId。后端要先持久化这个编号，并给它增加唯一约束；发送失败后的补发任务只能读取并复用这个已保存的编号，不能重新生成。
5.2 创建消费记录唯一约束
ACL 和 HRMS 属于不同 Group，所以同一事件应允许它们分别处理一次：
CREATE TABLE mq_consume_record (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
event_id VARCHAR(64) NOT NULL,
consumer_group VARCHAR(128) NOT NULL,
consumed_at DATETIME NOT NULL,
UNIQUE KEY uk_event_consumer (event_id, consumer_group)
);5.3 在同一事务中完成去重和业务更新
@Transactional(rollbackFor = Exception.class)
public void consumeIdempotently(AccountChangedEvent event, String consumerGroup) {
// insertIfAbsent：首次插入返回 1，唯一键已存在返回 0。
int inserted = consumeRecordRepository.insertIfAbsent(
event.getEventId(),
consumerGroup
);
if (inserted == 0) {
// 同一 Group 已成功处理过，直接结束；监听器返回 CommitMessage。
return;
}
// 消费记录和账户更新处于同一个事务。
// 更新失败时两者一起回滚，下次重投仍可以继续处理。
accountRepository.saveOrUpdate(event.getAccount());
}对应 SQL：
INSERT IGNORE INTO mq_consume_record(event_id, consumer_group, consumed_at)
VALUES (?, ?, NOW());5.4 可靠性验证清单
| 测试场景 | SOFAMQ 下的预期结果 |
|---|---|
同一 requestId 并发提交多次 | 账户只修改一次，只生成一个 eventId 和一条本地消息记录 |
producer.send() 抛异常 | 本地消息状态变为待重试，不只是打印日志 |
同一 eventId 投递两次 | ACL、HRMS 各自只产生一次业务效果 |
| ACL 数据库暂时不可用 | 返回 ReconsumeLater，恢复后重新消费 |
| 消息格式永久错误 | 保存坏消息并告警，返回 CommitMessage 停止无意义重试 |
同一 accId 连续变化 | 使用 OrderProducer 时按 ShardingKey 顺序消费，并由版本号防旧覆盖新 |
| Consumer 全部暂停 | Group 堆积量和延迟上升，恢复后继续消费 |
| 连续重试仍失败 | 进入对应 Group ID 的死信队列，可查询并重新发送 |
aboss-sso 成功但消息未发出 | 事务消息回查或本地消息表补发 |
六、🎤 面试回答模板
6.1 一分钟回答
我们项目使用 SOFAMQ。生产端通过同步
Producer.send()获取SendResult，失败不能只记日志，而要用本地消息表或事务消息补发。消费端处理成功返回Action.CommitMessage，临时失败返回Action.ReconsumeLater，超过重试次数后处理死信。由于异常重试可能造成重复，我们给每次业务事件生成eventId，通过消费记录唯一约束和业务事务保证幂等。如果同一账户必须有序，则使用OrderProducer，把accId作为ShardingKey。堆积时按 Group ID 查看在线状态、消费速度、堆积量和延迟，再定位慢 SQL、异常重试或消费能力不足。最后通过事务消息或本地消息表，加上对账补偿，保证aboss-sso、ACL 和 HRMS 最终一致。
6.2 常见追问
SOFAMQ 需要开启 Producer Confirm 吗？
不需要单独的 Confirm 开关。同步 producer.send(message) 正常返回 SendResult 表示取得发送结果，失败会抛异常；异步发送则通过 onSuccess 和 onException 回调判断。
为什么成功返回 SendResult 还不代表链路完成？
SendResult 只覆盖生产者到 SOFAMQ 服务端。ACL、HRMS 是否处理成功，要看各自 Group 的消费结果和业务数据。
为什么消费失败不能总是返回 CommitMessage？
CommitMessage 表示当前 Group 已经处理完成，SOFAMQ 不会自动重投。数据库临时故障等可恢复问题应该返回 ReconsumeLater。
怎样保证同一账户消息顺序？
使用 OrderProducer.send(message, accId)，把 accId 作为 ShardingKey，让同一账户进入同一顺序分区；消费者再用版本号防止旧数据覆盖新数据。
Message Key 和 ShardingKey 有什么区别？
Message Key 主要用于标识和查询消息轨迹；ShardingKey 用于顺序消息的分区选择。当前场景可用 eventId 作为 Message Key，用 accId 作为 ShardingKey。
七、📌 总结
- 发送：同步
SendResult或异步回调只是第一层，还要有失败补发。 - 消费：成功返回
CommitMessage，临时失败返回ReconsumeLater。 - 去重：一次账户变更一个
eventId，消费记录与业务更新放在同一事务。 - 顺序：需要时使用
OrderProducer，以accId作为ShardingKey。 - 堆积：按 Group ID 查看消费状态、堆积量和延迟，再定位消费者瓶颈。
- 一致：SOFAMQ 事务消息或本地消息表负责最终发出，死信、对账和补偿负责兜底。