如何解决微服务下引起的 分布式事务题目

一、什么是分布式事务？

固然叫分布式事务，但不是一定是分布式部署的服务之间才会产生分布式事务。不是在同一个服务或同一个数据库架构下，产生的事务，也就是分布式事务。

• 跨数据源的分布式事务
  
• 跨服务的分布式事务
  

二、解决方案

1、利用阿里开源的Seata框架解决分布式事务

​ 1）seata的架构
​ Seata事务管理中有三个重要的角色：

• TC (Transaction Coordinator) - **事务和谐者：**维护全局和分支事务的状态，和谐全局事务提交或回滚。
  
• TM (Transaction Manager) - **事务管理器：**定义全局事务的范围、开始全局事务、提交或回滚全局事务。
  
• RM (Resource Manager) - **资源管理器：**管理分支事务处理的资源，与TC交谈以注册分支事务和陈诉分支事务的状态，并驱动分支事务提交或回滚。
  
  
  
  3）Seata的常用模式利用
  
  
  
  • XA模式
    在一阶段各个本地事务执行完成后，不提交，把执行状态给事务和谐者TC,此时本地事务继续持有数据库锁
    二阶段TC基于一阶段的陈诉来举行判断，假如一阶段均成功则通知所有的事务参与者，提交事务，假如一阶段任意一个参与者失败，则通知所有事务参与者回滚事务。
    优点：可以或许实现强一致性，满足ACID原则；实现简朴
    缺点：性能较差；依赖数据库的事务
    I. 在application.yml文件中开启XA模式(所有参与事务的服务都需要设置)：
    
    1. seata:
    2.   data-source-proxy-mode: XA  

    复制代码
    II. 在全局事务的入口方法添加@GlobalTransactional注解
    
    2.     @Override
    3.     @GlobalTransactional
    4.     public Long create(Order order) {
    5.         // 创建订单
    6.         orderMapper.insert(order);
    7.         try {
    8.             // 扣用户余额
    9.             accountClient.deduct(order.getUserId(), order.getMoney());
    10.             // 扣库存
    11.             storageClient.deduct(order.getCommodityCode(), order.getCount());
    12.         } catch (FeignException e) {
    13.             log.error("下单失败，原因:{}", e.contentUTF8(), e);
    14.             throw new RuntimeException(e.contentUTF8(), e);
    15.         }
    16.         return order.getId();
    17.     }

    复制代码
  • AT模式
    和xa模式一样也是二阶段提交，不同的是AT模式本地事务竣事后，直接提交。但是，它会在本地事务举行数据库数据更新的时间记录一下更新前后的快照。
    
  
  

​ 在二阶段需要回滚的时间，根据快照举行数据的恢复，假如二阶段全局事务提交，则把记录的快照删除。
​ 优点：性能好；实现也较为简朴
​ 缺点： 存在中间状态，只能达到最终的一致性；快照功能会影响一些性能，但是相对于XA模式照旧要好很多
I. 在application.yml文件中开启AT模式(所有参与事务的服务都需要设置)：

1. seata:
2.   data-source-proxy-mode: AT # 默认就是AT

复制代码

II. 创建相关数据库表

1. #在分支事务所在的库里创建记录快照的表undo_log
2. DROP TABLE IF EXISTS `undo_log`;
3. CREATE TABLE `undo_log`  (
4.   `branch_id` bigint(20) NOT NULL COMMENT 'branch transaction id',
5.   `xid` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT 'global transaction id',
6.   `context` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT 'undo_log context,such as serialization',
7.   `rollback_info` longblob NOT NULL COMMENT 'rollback info',
8.   `log_status` int(11) NOT NULL COMMENT '0:normal status,1:defense status',
9.   `log_created` datetime(6) NOT NULL COMMENT 'create datetime',
10.   `log_modified` datetime(6) NOT NULL COMMENT 'modify datetime',
11.   UNIQUE INDEX `ux_undo_log`(`xid`, `branch_id`) USING BTREE
12. ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = 'AT transaction mode undo table' ROW_FORMAT = Compact;
14. #在TC服务所使用的库里创建全局锁记录表lock_table
15. DROP TABLE IF EXISTS `lock_table`;
16. CREATE TABLE `lock_table`  (
17.   `row_key` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
18.   `xid` varchar(96) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
19.   `transaction_id` bigint(20) NULL DEFAULT NULL,
20.   `branch_id` bigint(20) NOT NULL,
21.   `resource_id` varchar(256) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
22.   `table_name` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
23.   `pk` varchar(36) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
24.   `gmt_create` datetime NULL DEFAULT NULL,
25.   `gmt_modified` datetime NULL DEFAULT NULL,
26.   PRIMARY KEY (`row_key`) USING BTREE,
27.   INDEX `idx_branch_id`(`branch_id`) USING BTREE
28. ) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;

复制代码

III. 在全局事务的入口方法添加@GlobalTransactional注解

2.     @Override
3.     @GlobalTransactional
4.     public Long create(Order order) {
5.         // 创建订单
6.         orderMapper.insert(order);
7.         try {
8.             // 扣用户余额
9.             accountClient.deduct(order.getUserId(), order.getMoney());
10.             // 扣库存
11.             storageClient.deduct(order.getCommodityCode(), order.getCount());
12.         } catch (FeignException e) {
13.             log.error("下单失败，原因:{}", e.contentUTF8(), e);
14.             throw new RuntimeException(e.contentUTF8(), e);
15.         }
16.         return order.getId();
17.     }

复制代码

2、利用RocketMQ实现可靠消息最终一致性方案 （适用于不同项目的环境）

模仿转账 a银行向b银行转账
a银行业务代码：
减少金额，像mq发送事务消息

• 引入rocketmq依赖
  

1. <dependency>
2.    <groupId>org.apache.rocketmq</groupId>
3.    <artifactId>rocketmq‐spring‐boot‐starter</artifactId>
4.    <version>2.0.2</version>
5. </dependency>

复制代码

2）设置rocketmq

1. rocketmq.producer.group = zhuoye #设置生产者组的名称
2. rocketmq.name‐server = 127.0.0.1:9876  #指定rocketmq的地址

复制代码

3） 业务层代码

1. @Service
2. @Slf4j
3. public class UserAccountServiceImpl implements UserAccountService {
5.    @Autowired
6.    private UserAccountMapper userAccountMapper;
8.    @Autowired
9.    private  RocketMQTemplate rocketMQTemplate;
10.    
11.    @Autowired
12.    private TansactionalRecordMapper tansactionalRecordMapper;
15.     //向mq发送转账消息
16.     @Override
17.     public void sendTransferAccountsMessagesToMq(AccountChangeEvent accountChangeEvent) {
19.         //将accountChangeEvent转成json
20.         JSONObject jsonObject =new JSONObject();
21.         jsonObject.put("transferAccountInfo",accountChangeEvent);
22.         String jsonString = jsonObject.toJSONString();
23.         //生成message类型
24.         Message<String> message = MessageBuilder.withPayload(jsonString).build();
25.         //发送一条事务消息
26.         /**
27.          * String txProducerGroup 生产组
28.          * String destination topic，
29.          * Message<?> message, 消息内容
30.          * Object arg 参数
31.          */
32.         rocketMQTemplate.sendMessageInTransaction("transferAccount_ABank","topic_transferAccount",message,null);
34.     }
36.     //更新账户，扣减金额
37.     @Override
38.     @Transactional
39.     public void doUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {
40.         //幂等判断
41.         if(tansactionalRecordMapper.isExist(accountChangeEvent.getTxNo())>0){
42.             return ;
43.         }
44.         //扣减金额
45.         userAccountMapper.updateAccountBalance(accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount() * -1);
46.         //添加事务日志
47.         tansactionalRecordMapper.add(accountChangeEvent.getTxNo());
48.     }

复制代码

4）编写RocketMQLocalTransactionListener接口实现类

1. @Component
2. @Slf4j
3. @RocketMQTransactionListener(txProducerGroup = "transferAccount_ABank")
4. public class ProducerTxmsgListener implements RocketMQLocalTransactionListener {
6.     @Autowired
7.     private UserAccountService userAccountService;
9.    @Autowired
10.    private UserAccountMapper userAccountMapper;
11.    
12.    @Autowired
13.    private TansactionalRecordMapper tansactionalRecordMapper;
15.     //事务消息发送后的回调方法，当消息发送给mq成功，此方法被回调
16.     @Override
17.     @Transactional
18.     public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
20.         try {
21.             //解析message，转成AccountChangeEvent
22.             String messageString = new String((byte[]) message.getPayload());
23.             JSONObject jsonObject = JSONObject.parseObject(messageString);
24.             String accountChangeString = jsonObject.getString("transferAccountInfo");
25.             //将accountChange（json）转成AccountChangeEvent
26.             AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
27.             //执行本地事务，扣减金额
28.             userAccountService.doUpdateAccountBalance(accountChangeEvent);
29.             //当返回RocketMQLocalTransactionState.COMMIT，自动向mq发送commit消息，mq将消息的状态改为可消费
30.             return RocketMQLocalTransactionState.COMMIT;
31.         } catch (Exception e) {
32.             e.printStackTrace();
33.             return RocketMQLocalTransactionState.ROLLBACK;
34.         }
35.     }
37.     //事务状态回查，查询是否扣减金额
38.     @Override
39.     public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
40.         //解析message，转成AccountChangeEvent
41.         String messageString = new String((byte[]) message.getPayload());
42.         JSONObject jsonObject = JSONObject.parseObject(messageString);
43.         String accountChangeString = jsonObject.getString("transferAccountInfo");
44.         //将accountChange（json）转成AccountChangeEvent
45.         AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
46.         //事务id
47.         String txNo = accountChangeEvent.getTxNo();
48.         int isExist = tansactionalRecordMapper.isExist(txNo);
49.         if(isExist>0){
50.             return RocketMQLocalTransactionState.COMMIT;
51.         }else{
52.             return RocketMQLocalTransactionState.UNKNOWN;
53.         }
54.     }
55. }

复制代码

b银行业务代码(前两步一样)：
接收消息，增长金额
1）业务层代码

1. @Service
2. @Slf4j
3. public class UserAccountServiceImpl implements UserAccountService {
5.    @Autowired
6.    private UserAccountMapper userAccountMapper;
7.    
8.    @Autowired
9.    private TansactionalRecordMapper tansactionalRecordMapper;
11.     //更新账户，增加金额
12.     @Override
13.     @Transactional
14.     public void addAccountInfoBalance(AccountChangeEvent accountChangeEvent) {
15.         //已更新
16.         if(tansactionalRecordMapper.isExist(accountChangeEvent.getTxNo())>0){
17.             return ;
18.         }
19.         //增加金额
20.         userAccountMapper.updateAccountBalance(accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());
21.         //添加事务记录，用于幂等
22.         tansactionalRecordMapper.add(accountChangeEvent.getTxNo());
23.     }
24. }

复制代码

2)监听事务消息

1. @Component
2. @Slf4j
3. @RocketMQMessageListener(consumerGroup = "transferAccount_ABank",topic = "topic_transferAccount",maxReconsumeTimes = 3)
4. public class TxmsgConsumer implements RocketMQListener<String> {
6.     @Autowired
7.     UserAccountService userAccountService;
9.     //接收消息
10.     @Override
11.     public void onMessage(String message) {
12.         //解析消息
13.         JSONObject jsonObject = JSONObject.parseObject(message);
14.         String accountChangeString = jsonObject.getString("transferAccountInfo");
15.         //转成AccountChangeEvent
16.         AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
17.         //更新本地账户，增加金额
18.         userAccountService.addAccountInfoBalance(accountChangeEvent);
19.     }
20. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。