最大努力通知也是一种解决分布式事务的方案,下边是一个是充值的例子:
通过对最大努力通知的理解,采用MQ的ack机制就可以实现最大努力通知。
方案1 :
本实例通过RocketMq中间件实现最大努力通知型分布式事务,模拟充值过程。
本案例有账户系统和充值系统两个微服务,其中账户系统的数据库是bank1数据库,其中有张三账户。充值系统的数据库使用bank1_pay数据库,记录了账户的充值记录。
业务流程如下图 :
本示例程序组成部分如下 :
数据库:MySQL-5.7.25
包括bank1和bank1_pay两个数据库。
JDK:64位 jdk1.8.0_201
rocketmq 服务端:RocketMQ-4.5.0
rocketmq 客户端:RocketMQ-Spring-Boot-starter.2.0.2-RELEASE 微服务框架:spring-boot-2.1.3、spring-cloud-Greenwich.RELEASE
微服务及数据库的关系 :
dtx/dtx-notifymsg-demo/dtx-notifymsg-demo-bank1 银行1,操作张三账户, 连接数据库bank1 dtx/dtx-notifymsg-demo/dtx-notifymsg-demo-pay 银行2,操作充值记录,连接数据库bank1_pay
创建bank1库,并导入以下表结构和数据(包含张三账户)
CREATE DATABASE `bank1` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci';DROP TABLE IF EXISTS `account_info`; CREATE TABLE `account_info` (`id` bigint(20) NOT NULL AUTO_INCREMENT,`account_name` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '户 主姓名',`account_no` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '银行卡号',`account_password` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT'帐户密码',`account_balance` double NULL DEFAULT NULL COMMENT '帐户余额', PRIMARY KEY (`id`) USING BTREE) ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;INSERT INTO `account_info` VALUES (2, '张三的账户', '1', '', 10000);DROP TABLE IF EXISTS `de_duplication`; CREATE TABLE `de_duplication` (`tx_no` varchar(64) COLLATE utf8_bin NOT NULL, `create_time` datetime(0) NULL DEFAULT NULL, PRIMARY KEY (`tx_no`) USING BTREE) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;
创建bank1_pay库,并导入以下表结构:
CREATE DATABASE `bank1_pay` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci'; CREATE TABLE `account_pay` (`id` varchar(64) COLLATE utf8_bin NOT NULL,`account_no` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '账号', `pay_amount` double NULL DEFAULT NULL COMMENT '充值余额',`result` varchar(20) COLLATE utf8_bin DEFAULT NULL COMMENT '充值结果:success,fail',PRIMARY KEY (`id`) USING BTREE) ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;
rocketMQ启动方式与RocketMQ实现可靠消息最终一致性事务中完全一致
discover-server是服务注册中心,测试工程将自己注册至discover-server。
(1)父工程maven依赖说明
在dtx父工程中指定了SpringBoot和SpringCloud版本
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring‐boot‐dependencies</artifactId> <version>2.1.3.RELEASE</version><type>pom</type><scope>import</scope></dependency><dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring‐cloud‐dependencies</artifactId> <version>Greenwich.RELEASE</version> <type>pom</type><scope>import</scope></dependency>
在dtx-notifymsg-demo父工程中指定了rocketmq-spring-boot-starter的版本。
<dependency><groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq‐spring‐boot‐starter</artifactId> <version>2.0.2</version></dependency>
(2)配置rocketMQ
在application-local.properties中配置rocketMQ nameServer地址及生产组 :
rocketmq.producer.group = producer_bank2 rocketmq.name-server = 127.0.0.1:9876
dtx-notifydemo-pay实现如下功能 :
1、充值接口;
2、充值完成要通知;
3、充值结果查询接口。
(2)Dao
@Mapper@Componentpublic 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); }
(3)Service
@Service@Slf4jpublic class AccountPayServiceImpl implements AccountPayService{@AutowiredRocketMQTemplate rocketMQTemplate;@AutowiredAccountPayDao accountPayDao;@Transactional@Overridepublic AccountPay insertAccountPay(AccountPay accountPay) {int result = accountPayDao.insertAccountPay(accountPay.getId(),accountPay.getAccountNo(), accountPay.getPayAmount(), "success");if(result>0){ //发送通知rocketMQTemplate.convertAndSend("topic_notifymsg",accountPay);return accountPay; }return null; }@Overridepublic AccountPay getAccountPay(String txNo) {AccountPay accountPay = accountPayDao.findByIdTxNo(txNo); return accountPay; }}
(4)Controller
@RestControllerpublic class AccountPayController {@AutowiredAccountPayService accountPayService;//充值@GetMapping(value = "/paydo")public AccountPay pay(AccountPay accountPay){//事务号String txNo = UUID.randomUUID().toString(); accountPay.setId(txNo);return accountPayService.insertAccountPay(accountPay);}//查询充值结果@GetMapping(value = "/payresult/{txNo}")public AccountPay payresult(@PathVariable("txNo") String txNo){return accountPayService.getAccountPay(txNo); }}
dtx-notifydemo-bank1实现如下功能 :
1、监听MQ,接收充值结果,根据充值结果完成账户金额修改。
2、主动查询充值系统,根据充值结果完成账户金额修改。
1)Dao
@Mapper@Componentpublic 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);}
2)AccountInfoService
@Service@Slf4jpublic class AccountInfoServiceImpl implements AccountInfoService {@AutowiredAccountInfoDao accountInfoDao;@AutowiredPayClient payClient; /** * 更新账号余额,并发送消息 * * @param accountChange */@Transactional@Overridepublic void updateAccountBalance(AccountChangeEvent accountChange) {//幂等校验int existTx = accountInfoDao.isExistTx(accountChange.getTxNo()); if(existTx >0){log.info("已处理消息:{}", JSONObject.toJSONString(accountChange));return ; }//添加事务记录 accountInfoDao.addTx(accountChange.getTxNo()); //更新账户金额accountInfoDao.updateAccountBalance(accountChange.getAccountNo(),accountChange.getAmount()); }/** * 主动查询充值结果 * * @param tx_no */@Overridepublic AccountPay queryPayResult(String tx_no) {//主动请求充值系统查询充值结果AccountPay accountPay = payClient.queryPayResult(tx_no); //充值结果String result = accountPay.getResult(); log.info("主动查询充值结果:{}", JSON.toJSONString(accountPay)); if("success".equals(result)){AccountChangeEvent accountChangeEvent = new AccountChangeEvent();accountChangeEvent.setAccountNo(accountPay.getAccountNo());accountChangeEvent.setAmount(accountPay.getPayAmount());accountChangeEvent.setTxNo(accountPay.getId());updateAccountBalance(accountChangeEvent);}return accountPay; }}
@FeignClient(value = "dtx‐notifymsg‐demo‐pay", fallback = PayFallback.class)public interface PayClient {@GetMapping("/pay/payresult/{txNo}")AccountPay queryPayResult(@PathVariable("txNo") String txNo); }@Componentpublic class PayFallback implements PayClient {@Overridepublic AccountPay queryPayResult(String txNo) {AccountPay accountPay = new AccountPay();accountPay.setResult("fail");return accountPay;} }
3)监听MQ
@Component@Slf4j@RocketMQMessageListener(topic="topic_notifymsg",consumerGroup="consumer_group_notifymsg_bank1") public class NotifyMsgListener implements RocketMQListener<AccountPay> {@AutowiredAccountInfoService accountInfoService;@Overridepublic void onMessage(AccountPay accountPay) {log.info("接收到消息:{}", JSON.toJSONString(accountPay)); AccountChangeEvent accountChangeEvent = new AccountChangeEvent();accountChangeEvent.setAmount(accountPay.getPayAmount());accountChangeEvent.setAccountNo(accountPay.getAccountNo());accountChangeEvent.setTxNo(accountPay.getId());accountInfoService.updateAccountBalance(accountChangeEvent); log.info("处理消息完成:{}", JSON.toJSONString(accountChangeEvent));} }
4)Controller
@RestController@Slf4jpublic class AccountInfoController {@Autowiredprivate AccountInfoService accountInfoService;//主动查询充值结果@GetMapping(value = "/payresult/{txNo}")public AccountPay result(@PathVariable("txNo") String txNo){AccountPay accountPay = accountInfoService.queryPayResult(txNo);return accountPay; }}
充值系统充值成功,账户系统主动查询充值结果,修改账户金额。
充值系统充值成功,发送消息,账户系统接收消息,修改账户金额。
账户系统修改账户金额幂等测试。
最大努力通知方案是分布式事务中对一致性要求最低的一种,适用于一些最终一致性时间敏感度低的业务;
最大努力通知方案需要实现如下功能 :
1、消息重复通知机制。
2、消息校对机制。
联系客服