分布式事务解决方案之最大努力通知（五）

博客参考bilibili视频：分布式事务解决方案专题

一 什么是最大努力通知

最大努力通知也是一种解决分布式事务的方案，下边是一个是充值的例子：

　　

1）交互流程:

1、账户系统调用充值系统接口；

2、充值系统完成支付处理向账户系统发起充值结果通知。

若通知失败，则充值系统按策略进行重复通知

3、账户系统接收到充值结果通知修改充值状态；

4、账户系统未接收到通知会主动调用充值系统的接口查询充值结果。

2）通过上边的例子我们总结最大努力通知方案的目标：

目标：发起通知方【支付系统服务】通过一定的机制最大努力将业务处理结果通知到接收方【账户系统服务】。具体包括：

1、有一定的消息重复通知机制；

　　因为接收通知方可能没有接收到通知，此时要有一定的机制对消息重复通知。

2、消息校对机制；

　　如果尽最大努力也没有通知到接收方，或者接收方消费消息后要再次消费，此时可由接收方主动向通知方查询消息   信息来满足需求。

3）最大努力通知与可靠消息一致性有什么不同？

1、解决方案思想不同

　　可靠消息一致性，发起通知方【张三转账】需要保证将消息发出去，并且将消息发到接收通知方【李四收账】，消息的可靠性关键由发起通知方【张三转账】来保证；

　　最大努力通知，发起通知方【支付系统服务】尽最大的努力将业务处理结果通知为接收通知方【账户系统服务】，但是可能消息接收不到，此时需要接收通知方【账户系统服务】主动调用发起通知方【支付系统服务】的接口查询业务处理结果，通知的可靠性关键在接收通知方【账户系统服务】。

2、两者的业务应用场景不同

【张三给李四转账，注重的交易过程，可靠消息最终一致性】

可靠消息一致性关注的是交易过程的事务一致，以异步的方式完成交易。    

【注册用户送积分，用户注册成功，增加积分成功后通知用户服务，或者提供用户服务查询是否注册成功的积分接口】

最大努力通知关注的是交易后的通知事务，即将交易结果可靠的通知出去。   

3、技术解决方向不同

　　可靠消息一致性要解决消息从发出到接收的一致性，即消息发出并且被接收到，接收不到需要消息重复发送，或者人工干预确保接收方完成消费【幂等性操作】；【确保消息发送的可靠性】

　　最大努力通知无法保证消息从发出到接收的一致性，只提供消息接收的可靠性机制。可靠机制是，最大努力的将消息通知给接收方，当消息无法被接收方接收时，由接收方主动查询消息（业务处理结果）。【确保消息接收的可靠性】

二 解决方案

通过对最大努力通知的理解，采用 MQ 的 ack 机制就可以实现最大努力通知。  

方案1：

【内部系统通信适用】

　　

本方案是利用 MQ 的 ack 机制由 MQ 向接收通知方发送通知，流程如下：

1、发起通知方将通知发给 MQ Server；

　　使用普通消息机制将通知发给 MQ Server。

注意：如果消息没有发出去可由接收通知方主动请求发起通知方查询业务执行结果。

2、接收通知方监听 MQ Server；

3、接收通知方接收消息，业务处理完成回应 ack；

4、接收通知方若没有回应 ack 则 MQ 会重复通知。

MQ会按照间隔 1min、5min、10min、30min、1h、2h、5h、10h 的方式，逐步拉大通知间隔 （如果MQ采用 RocketMq，在broker中可进行配置），直到达到通知要求的时间窗口上限。

5、接收通知方可通过消息校对接口来校对消息的一致性。

方案2：【外部系统通信适用】

本方案也是利用 MQ 的 ack 机制，与方案1不同的是应用程序向接收通知方发送通知，如下图：

 　　

交互流程如下：

1、发起通知方将通知发给MQ。

　　使用可靠消息一致性方案中的事务消息保证本地事务与消息的原子性，最终将通知先发给 MQ；

2、通知程序监听 MQ，接收 MQ 的消息。

　　方案1 中接收通知方直接监听 MQ，方案2 中由通知程序监听 MQ。

　　通知程序若没有回应 ack 则 MQ 会重复通知。

3、通知程序通过互联网接口协议（如 http、webservice）调用接收通知方接口，完成通知。

　　通知程序调用接收通知方接口成功就表示通知成功，即消费 MQ 消息成功，MQ 将不再向通知程序投递通知消息。

4、接收通知方可通过消息校对接口来校对消息的一致性。

方案1 和方案2 的不同点：

1、方案1 中接收通知方与 MQ 接口，即接收通知方监听 MQ，此方案主要应用与内部应用之间的通知；

2、方案2 中通知程序与 MQ 接口，通知程序监听 MQ，收到 MQ 的消息后由通知程序通过互联网接口协议调用接收通知方。此方案主要应用于外部应用之间的通知，例如支付宝、微信的支付结果通知。

三 RocketMQ实现最大努力通知型事务

1）业务说明

本实例通过 RocketMq 中间件实现最大努力通知型分布式事务，模拟充值过程；

本案例有账户系统和充值系统两个微服务，其中账户系统的数据库是 bank1 数据库，其中有张三账户。充值系统的数据库使用 bank1_pay 数据库，记录了账户的充值记录。

业务流程如下图：

　　　

交互流程如下： 

1、用户请求充值系统进行充值；

2、充值系统完成充值将充值结果发给 MQ；

3、账户系统监听 MQ，接收充值结果通知，如果接收不到消息，MQ 会重复发送通知，接收到充值结果通知账户系统增加充值金额；

4、账户系统也可以主动查询充值系统的充值结果查询接口，增加金额。

2）dtx-notifydemo-pay

dtx-notifydemo-pay 实现如下功能：  

1、充值接口

2、充值完成要通知

3、充值结果查询接口

a）Dao 

@Mapper
@Component
public interface AccountPayDao {

    @Insert("insert into account_pay(id,account_no,pay_amount,result) values(#{id},#{accountNo},#{payAmount},#{result})")
    int insertAccountPay(@Param("id") String id, @Param("accountNo") String accountNo, @Param("payAmount") Double pay_amount, @Param("result") String result);

    @Select("select id,account_no accountNo,pay_amount payAmount,result from account_pay where id=#{txNo}")
    AccountPay findByIdTxNo(@Param("txNo") String txNo);

}

b）Service 需提供插入充值记录，查询充值记录接口

@Service
@Slf4j
public class AccountPayServiceImpl implements AccountPayService {

    @Autowired
    private AccountPayDao accountPayDao;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 插入充值记录
    @Override
    public AccountPay insertAccountPay(AccountPay accountPay) {
        int success = accountPayDao.insertAccountPay(accountPay.getId(), accountPay.getAccountNo(), accountPay.getPayAmount(), "success");
        if (success > 0) {      // 插入成功发送消息
            // 发送通知，使用普通消息发送通知【需要指定 topic 主题】
            accountPay.setResult("result");     // 发送的消息 result 必须为 success
            rocketMQTemplate.convertAndSend("topic_notifymsg", accountPay);
            return accountPay;
        }
        return null;
    }

    // 查询充值记录，接收通知放调用此方法来查询充值结果
    @Override
    public AccountPay getAccountPay(String txNo) {
        AccountPay accountPay = accountPayDao.findByIdTxNo(txNo);
        return accountPay;
    }
}

c）controller 

@RestController
@Slf4j
public class AccountPayController {

    @Autowired
    private AccountPayService accountPayService;

    // 充值
    @GetMapping(value = "/paydo")
    public AccountPay pay(AccountPay accountPay) {
        // 生成事务号
        String txNo = UUID.randomUUID().toString();
        accountPay.setId(txNo);
        return accountPayService.insertAccountPay(accountPay);
    }

    // 远程调用充值系统的接口查询充值结果
    @GetMapping("/payresult/{txNo}")
    public AccountPay payresult(@PathVariable("txNo") String txNo) {
        return accountPayService.getAccountPay(txNo);
    }

}

3）dtx-notifydemo-bank1

dtx-notifydemo-bank1 实现如下功能：

1、监听 MQ，接收充值结果，根据充值结果完成账户金额修改；

2、主动查询充值系统，根据充值结果完成账户金额修改。

a）Dao 

@Mapper
@Component
public interface AccountInfoDao {
    //修改账户金额
    @Update("update account_info set account_balance=account_balance+#{amount} where account_no=#{accountNo}")
    int updateAccountBalance(@Param("accountNo") String accountNo, @Param("amount") Double amount);

   //查询幂等记录，用于幂等控制【账户接收通知可能会重复接收通知，需要做幂等操作】
    @Select("select count(1) from de_duplication where tx_no = #{txNo}")
    int isExistTx(String txNo);

    //添加事务记录，用于幂等控制【账户接收通知可能会重复接收通知，需要做幂等操作】
    @Insert("insert into de_duplication values(#{txNo},now());")
    int addTx(String txNo);

}

b）Service 更新账户金额、远程调用查询充值结果

@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {

    @Autowired
    private AccountInfoDao accountInfoDao;
    @Autowired
    private PayClient payClient;

    // 更新账户金额
    @Override
    @Transactional
    public void updateAccountBalance(AccountChangeEvent accountChangeEvent) {
        // 幂等性校验
        if (accountInfoDao.isExistTx(accountChangeEvent.getTxNo()) > 0) {
            log.info("幂等性校验，防止账户接收方重复消费消息");
            return;
        }
        accountInfoDao.updateAccountBalance(accountChangeEvent.getAccountNo(), accountChangeEvent.getAmount());

        // 插入事务记录，用户幂等控制
        accountInfoDao.addTx(accountChangeEvent.getTxNo());
    }

    // 远程调用查询充值结果
    @Override
    public AccountPay queryPayResult(String txNo) {
        // 远程调用【使用 openFeign 组件远程获取充值结果】
        AccountPay accountPay = payClient.payresult(txNo);

        // 成功获取充值用户金额数据，执行更新账户操作
        if ("success".equalsIgnoreCase(accountPay.getResult())) {
            // 更新账户金额
            AccountChangeEvent accountChangeEvent = new AccountChangeEvent();
            accountChangeEvent.setAccountNo(accountPay.getAccountNo()); // 充值账号
            accountChangeEvent.setAmount(accountPay.getPayAmount());    // 金额
            accountChangeEvent.setTxNo(accountPay.getId());     // 充值事务号
            this.updateAccountBalance(accountChangeEvent);
        }

        return accountPay;
    }
}

c）通过 openfeign 远程调用充值结果 

@FeignClient(value = "dtx-notifymsg-demo-pay", fallback = PayFallback.class)
public interface PayClient {

    // 远程调用充值系统的接口查询充值结果
    @GetMapping("/payresult/{txNo}")
    AccountPay payresult(@PathVariable("txNo") String txNo);

}

@Component
public class PayFallback implements PayClient {

　　 // 服务降级 
    @Override
    public AccountPay payresult(String txNo) {
        AccountPay accountPay = new AccountPay();
        accountPay.setResult("fail");
        return accountPay;
    }
}

d）监听 MQ

// 需要指定接收消息的 topic【充值提供方设置的 topic 】当服务接收方监听收不到消息时，consumerGroup 就是MQ重复发送消息的唯一标识【服务名】
@RocketMQMessageListener(topic = "topic_notifymsg", consumerGroup = "consumer_group_notifymsg_bank1")
@Component
@Slf4j
public class NotifyMsgListener implements RocketMQListener<AccountPay> {

    @Autowired
    private AccountInfoService accountInfoService;

    // 接收消息
    @Override
    public void onMessage(AccountPay accountPay) {
        log.info("接收到消息：{}", JSON.toJSON(accountPay));
        if ("success".equalsIgnoreCase(accountPay.getResult())) {
            // 更新账户金额
            AccountChangeEvent accountChangeEvent = new AccountChangeEvent();
            accountChangeEvent.setAccountNo(accountPay.getAccountNo());
            accountChangeEvent.setAmount(accountPay.getPayAmount());
            accountChangeEvent.setTxNo(accountPay.getId());
            accountInfoService.updateAccountBalance(accountChangeEvent);
        }
        log.info("处理完消息：{}", JSON.toJSON(accountPay));
    }

}

e）controller 账户系统主动查询充值系统的充值结果

@RestController
@Slf4j
public class AccountInfoController {

    @Autowired
    private AccountInfoService accountInfoService;

    // 账户系统主动查询充值系统的充值结果
    @GetMapping("/payresult/{txNo}")
    public AccountPay result(@PathVariable("txNo") String txNo) {
        AccountPay accountPay = accountInfoService.queryPayResult(txNo);
        return accountPay;
    }

}

四 小结

最大努力通知方案是分布式事务中对一致性要求最低的一种，适用于一些最终一致性时间敏感度低的业务，最大努力通知方案需要实现如下功能：

1、消息重复通知机制；

2、消息校对机制【事务回查机制】

 　　

　　每天进步一点点......