4. SOFAMQ 账户变更代码实践

一、💡 一句话理解

核心结论

aboss-sso 把最新账户详情发送到 SOFAMQ；ACL 和 HRMS 使用不同 Consumer Group 订阅同一 Topic，因此两个系统都能处理这条消息，同组多个实例再共同分担。

前置概念：

• 2. MQ 核心角色与代码对应
• 3. MQ Consumer Group 与消费模式

二、🧭 完整调用链

ExternalAccountUpdateCmdExe.execute()
    ↓ 修改账户数据
AbsAccountChangeCmdExe.publish(accId)
    ↓ 查询最新 AccountDetailDTO
AccountChangeMessage.publish(dto)
    ↓ 创建 Message
SOFAMQ Producer.send(message)
    ↓
TP_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY
    ├── ACL Group  → 新增或更新 ACL 账户
    └── HRMS Group → 处理 HRMS 账户变化

三、⚙️ 生产端配置

# 是否创建普通消息 Producer。
spring.sofamq.normal.producers.enable=true
 
# 连接 MQ 使用的身份、实例和接入地址。
# 真实值位于各环境配置文件中，不能写进学习笔记。
spring.sofamq.normal.access-key=***
spring.sofamq.normal.secret-key=***
spring.sofamq.normal.instance-id=***
spring.sofamq.normal.endpoint=***
 
# 发送代码通过这个名称选择 Producer。
spring.sofamq.normal.producers[1].name=accountChangeMessage
 
# 账户变更消息的业务分类。
spring.sofamq.account.change.topic=TP_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY
 
# Topic 内进一步过滤的标签。
spring.sofamq.account.change.tag=TG_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY

主要代码位置：

aboss-sso-support-app/.../executor/account/AbsAccountChangeCmdExe.java
aboss-sso-support-infrastructure/.../mq/AccountChangeMessage.java
aboss-sso-support-infrastructure/.../config/ProducerConfig.java
start/src/main/resources/application.properties

四、📤 生产者怎样发送消息

4.1 业务执行器触发发布

@Transactional(rollbackFor = Exception.class)
public String execute(ExternalAccountUpdateCmd cmd) {
    Account account = init(cmd);
 
    update(account);                 // 更新 aboss-sso 账户数据。
    modify(account);                 // 同步修改 IDaaS 账户。
    buildRelatedAccount(account);    // 维护关联账户关系。
    publish(account.getAccountId()); // 发布账户变更消息。
 
    return account.getAccountId();
}

4.2 查询变更后的完整数据

protected void publish(String accId) {
    // 根据账户 ID 查询最新账户详情。
    List<AccountDetailDTO> details =
            accountListByAccountIdsQryExe.execute(
                    new AccountIdsQry()
                            .setAccIds(ImmutableList.of(accId))
            );
 
    if (CollectionUtils.isNotEmpty(details)) {
        AccountDetailDTO detail = details.get(0);
 
        // 离职账户发送原手机号，方便下游识别原账户。
        if (!detail.getIsValid()) {
            detail.setPhoneNo(detail.getBeforePhoneNo())
                    .setBeforePhoneNo(null);
        }
 
        // 把最新账户详情交给消息发送器。
        accountChangeMessage.publish(detail);
 
        // 清理账户详情缓存。
        accountGateway.cleanAccountDetailCache(accId);
    }
}

4.3 创建 Message 并发送

@Override
public void publish(AccountDetailDTO accountDetailDTO) {
    Message message = new Message(
            topic,
            tag,
            JSON.toJSONString(accountDetailDTO)
                    .getBytes(StandardCharsets.UTF_8)
    );
 
    try {
        getProducer().send(message);
    } catch (Exception e) {
        log.error(
                "account info modify mq error, body: {}",
                JSON.toJSONString(accountDetailDTO),
                e
        );
    }
}

getProducer() 的真实实现是按配置名称从 ProducerSelector 中取出 Producer：

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

概念	项目代码
业务数据	AccountDetailDTO
Message Body	DTO 序列化后的 JSON 字节数组
Topic	TP_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY
Tag	TG_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY
Producer	getProducer() 返回的对象
发送	send(message)

五、📥 ACL 怎样消费消息

5.1 ACL 消费配置

spring.sofamq.normal.consumers[7].group-id=GID_CIC_MIDABOSS_ACL_ACCOUNT_MODIFY
spring.sofamq.normal.consumers[7].bean-name=accountCenterListener
spring.sofamq.normal.consumers[7].topic=TP_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY
spring.sofamq.normal.consumers[7].tag=TG_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY

5.2 ACL 消费代码

下面是根据真实代码压缩后的学习版，validate 和 convert 用来代表原代码中的详细校验与字段转换：

// 根据配置创建一个属于 ACL Group 的 Consumer。
Consumer consumer = accessPoint.createConsumer(properties);
 
// 订阅与生产端一致的 Topic 和 Tag。
consumer.subscribe(topic, tag, new MessageListener() {
    @Override
    public Action consume(Message message, ConsumeContext context) {
        // 1. 将消息体从字节数组恢复成 JSON。
        String body = new String(
                message.getBody(),
                StandardCharsets.UTF_8
        );
 
        // 2. 转换成 ACL 使用的请求 DTO。
        SsoAccountDetailReqDTO account =
                JSONUtils.parseObject(body, SsoAccountDetailReqDTO.class);
 
        // 3. 校验账户 ID、登录名、组织等字段。
        validate(account);
 
        // 4. 按 accId 判断新增还是更新。
        AccountDetailResModel old =
                accountService.getAccountDetailByAccId(account.getAccId());
 
        if (old != null) {
            accountService.updateAccount(convert(account));
        } else {
            accountService.insertAccount(convert(account));
        }
 
        // 5. 业务完成后确认成功。
        return Action.CommitMessage;
    }
});
 
// 启动 Consumer，开始获取消息。
consumer.start();

这段代码的主线是：

订阅 → 读取 Body → 反序列化 → 校验 → 新增或更新 → Commit

六、👥 HRMS 怎样集群消费

HRMS 使用另一个 Group ID：

spring.sofamq.account.outsourced.group=GID_CIC_MIDABOSS_HRMS_ACCOUNT_OUTSOURCED_CHANG_STAFF
spring.sofamq.account.outsourced.change.topic=TP_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY
spring.sofamq.account.outsourced.change.tag=TG_CIC_MIDABOSS_SSO_ACCOUNT_MODIFY

消费端明确设置集群模式：

properties.setProperty(PropertyKeyConst.GROUP_ID, groupId);
 
// 同一个 HRMS Group 的多个实例共同分担消息。
properties.setProperty(
        PropertyKeyConst.MESSAGE_MODEL,
        PropertyValueConst.CLUSTERING
);
 
Consumer consumer = accessPoint.createConsumer(properties);
 
consumer.subscribe(topic, accountChangedTag, (message, context) -> {
    String body = new String(
            message.getBody(),
            StandardCharsets.UTF_8
    );
 
    AccountDetailDTO account =
            JSONUtils.parseObject(body, AccountDetailDTO.class);
 
    // 此处省略真实业务分支。
    handle(account);
 
    return Action.CommitMessage;
});
 
consumer.start();

ACL 和 HRMS 的 Topic 相同，但 Group ID 不同，因此两个系统各自消费一遍。HRMS 如果部署三个实例，再由三个实例共同分担 HRMS Group 的消息。

七、🚨 消费失败怎样返回

这里的“返回”是什么意思

consume() 不是 HTTP 接口，所以不会向前端或生产者返回 JSON、错误码。它只把一个 Action 枚举值返回给调用它的 MQ 客户端 SDK，SDK 再根据这个值告诉 Broker：确认成功，或者稍后重新投递。

7.1 返回给谁

consume() 是 MQ 客户端 SDK 调用的回调方法：

Broker 把消息交给 Consumer SDK
        ↓
SDK 调用 consume(message, context)
        ↓
业务代码处理消息
        ↓
consume() 返回 Action 枚举值给 SDK
        ↓
SDK 向 Broker 确认成功，或者要求稍后重试

生产者 aboss-sso 不会收到这个返回值。普通 MQ 消息是异步通信；如果生产者必须知道下游业务处理结果，需要另外设计回执消息、状态表或查询接口。

7.2 具体返回什么

处理情况	Java 返回值	MQ 后续动作
业务处理成功	Action.CommitMessage	确认消费成功，这条消息不再投递给当前 Group
数据库暂时不可用、调用下游超时等临时故障	Action.ReconsumeLater	本次消费失败，Broker 稍后把消息重新投递给当前 Group
消息字段缺失、格式非法等永久性脏数据	保存原消息并告警后返回 Action.CommitMessage	停止重试，避免同一条坏消息不断失败

所以，“消费失败具体返回什么”的直接答案是：

return Action.ReconsumeLater;

它返回的不是下面这种业务响应：

{
  "code": 500,
  "message": "消费失败"
}

而是 com.aliyun.openservices.ons.api.Action 中的一个枚举值，由 MQ SDK 解释。

7.3 放进消费代码里怎样写

public Action consume(Message message, ConsumeContext context) {
    try {
        AccountDetailDTO account = parseAndValidate(message);
 
        // 业务必须支持重复执行。
        accountService.saveOrUpdate(account);
 
        // 返回给 MQ SDK：业务已经处理成功，不要再投递这条消息。
        return Action.CommitMessage;
    } catch (RetryableException e) {
        // 返回给 MQ SDK：本次没有处理成功，请让 Broker 稍后重新投递。
        // 适用于数据库临时不可用、下游超时等可恢复故障。
        return Action.ReconsumeLater;
    } catch (IllegalArgumentException e) {
        // 永久性脏数据：保存原消息并告警，避免无限重试。
        saveBadMessageAndAlarm(message, e);
 
        // 虽然本次业务没有成功，但继续重试也不会恢复，
        // 因此确认这条消息，阻止它反复投递。
        return Action.CommitMessage;
    }
}

官方文档还说明，TCP 客户端返回 null 或直接抛出异常也会触发重试。不过显式返回 Action.ReconsumeLater 更容易让人看懂代码意图，也方便区分“可恢复故障”和“永久性脏数据”。

返回 CommitMessage 不等于业务成功

它准确表达的是“MQ 不要再投递这条消息”。所以处理永久性脏数据时，可以先保存原消息并告警，再返回 CommitMessage。如果直接确认但没有留下记录，消息就可能静默丢失。

其中 parseAndValidate、saveOrUpdate、saveBadMessageAndAlarm 和异常类型是用于解释流程的占位内容，不代表项目一定存在同名实现。

记忆口诀：CommitMessage 表示“处理结束，不再发送”；ReconsumeLater 表示“这次没成，稍后再发”。

更完整的可靠性处理见 5. MQ 可靠性问题 和 6. 消息幂等。

八、🔍 怎样验证完整链路

1. 在测试环境修改一个外部账户。
2. 在执行器和 AccountChangeMessage.publish 设置断点。
3. 确认 AccountDetailDTO 是数据库更新后的数据。
4. 在 MQ 控制台按 Topic、Tag 或消息 ID 查询发送结果。
5. 分别检查 ACL Group 和 HRMS Group 的消费进度与堆积量。
6. 核对 ACL、HRMS 的最终业务数据。
7. 重复投递同一业务消息，验证消费逻辑是否幂等。

当前代码的可靠性风险

AccountChangeMessage 捕获发送异常后只记录日志，没有继续抛出。MQ 发送失败时，外层事务不一定回滚，接口也可能继续返回成功；反过来，消息发送成功后数据库事务仍可能提交失败。重要链路还需要发送确认、失败补偿、事务消息或本地消息表，并让消费者保持幂等。

九、🎤 面试时怎么说

aboss-sso 在账户变更完成后查询最新的 AccountDetailDTO，将它序列化成 JSON，并通过 SOFAMQ/OpenMessaging Producer 发送到账户变更 Topic。ACL 和 HRMS 使用不同 Group ID 订阅，所以两个系统都能处理这条消息；HRMS 使用集群模式，同组多个实例共同分担。ACL 消费后按 accId 新增或更新本地账户，并在业务成功后返回 CommitMessage。这条链路实现了解耦，但生产端异常只记日志、消费端又可能重复收到消息，因此还需要发送补偿、消费幂等和监控告警。

十、📌 快速回顾

• aboss-sso 是 Producer，ACL 和 HRMS 是 Consumer。
• Topic 和 Tag 决定发送与订阅范围。
• 不同 Group 各自消费，同组多个实例共同分担。
• 业务成功后才返回 CommitMessage。
• 可恢复故障可以返回 ReconsumeLater。
• 发送只记日志是当前链路需要关注的可靠性风险。

十一、🔗 相关资料

• ../../private/对外接口文档/账号中心/账户变更对外广播消息文档
• 5. MQ 可靠性问题
• 6. 消息幂等
• 阿里云 RocketMQ：消息重试
• Apache RocketMQ 5.0：Consumer Group
• Apache RocketMQ 5.0：Consumption Retry