LCN分布式事务管理

Posted 2020-11-24 fengyuduke

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了LCN分布式事务管理相关的知识，希望对你有一定的参考价值。

前言

好久没写东西了，9月份换了份工作，一上来就忙的要死。根本没时间学东西，好在新公司的新项目里面遇到了之前没遇到过的难题。那遇到难题就要想办法解决咯，一个请求，调用两个服务，同时操作更新两个数据库。这就带来事务不一致的问题了，分布式事务管理被强行拉出来了。导致原本两个springboot的单体项目，必须要协同管理起来。刚好微服务也接触过，小试牛刀咯。

框架介绍

LCN分布式事务框架其本身并不创建事务，而是基于对本地事务的协调从而达到事务一致性的效果

核心步骤

创建事务组
是指在事务发起方开始执行业务代码之前先调用TxManager创建事务组对象，然后拿到事务标示GroupId的过程。

添加事务组
添加事务组是指参与方在执行完业务方法以后，将该模块的事务信息添加通知给TxManager的操作。

关闭事务组
是指在发起方执行完业务代码以后，将发起方执行结果状态通知给TxManager的动作。当执行完关闭事务组的方法以后，TxManager将根据事务组信息来通知相应的参与模块提交或回滚事务。

框架的特点

支持各种基于spring的db框架
兼容SpringCloud、Dubbo、motan
使用简单，低依赖，代码完全开源
基于切面的强一致性事务框架
高可用，模块可以依赖RPC模块做集群化，TxManager也可以做集群化
支持本地事务和分布式事务共存
支持事务补偿机制，增加事务补偿决策提醒
添加插件拓展机制

源码目录说明

transaction-dubbo LCN dubbo rpc框架扩展支持
transaction-springcloud LCN springcloud rpc框架扩展支持
tx-client 是LCN核心tx模块端控制框架
tx-manager 是LCN 分布式事务协调器
tx-plugins-db 是LCN 对关系型数据库的插件支持
tx-plugins-nodb 是LCN 对于无数据库模块的插件支持

地址：https://gitee.com/fengyuduke/my_open_resources/blob/master/tx-lcn-master.7z

本章只说微服务模式下的使用。

1.配置分布式事务协调器

只要配置三个地方：redis连接信息，eureka注册中心，开放的端口。

eureka注册中心一定要配置自己系统的地址。

#######################################txmanager-start#################################################
#服务端口
server.port=8899

#tx-manager不得修改
spring.application.name=tx-manager

spring.mvc.static-path-pattern=/**
spring.resources.static-locations=classpath:/static/
#######################################txmanager-end#################################################


#zookeeper地址
#spring.cloud.zookeeper.connect-string=127.0.0.1:2181
#spring.cloud.zookeeper.discovery.preferIpAddress = true

#eureka 地址
eureka.client.service-url.defaultZone=http://localhost:8083/eureka/
eureka.instance.prefer-ip-address=true

#######################################redis-start#################################################
#redis 配置文件，根据情况选择集群或者单机模式

##redis 集群环境配置
##redis cluster
#spring.redis.cluster.nodes=127.0.0.1:7001,127.0.0.1:7002,127.0.0.1:7003
#spring.redis.cluster.commandTimeout=5000

##redis 单点环境配置
#redis
#redis主机地址
spring.redis.host=127.0.0.1
#redis主机端口
spring.redis.port=6379
#redis链接密码
spring.redis.password=
spring.redis.pool.maxActive=10
spring.redis.pool.maxWait=-1
spring.redis.pool.maxIdle=5
spring.redis.pool.minIdle=0
spring.redis.timeout=0
#####################################redis-end###################################################




#######################################LCN-start#################################################
#业务模块与TxManager之间通讯的最大等待时间（单位：秒）
#通讯时间是指：发起方与响应方之间完成一次的通讯时间。
#该字段代表的是Tx-Client模块与TxManager模块之间的最大通讯时间，超过该时间未响应本次请求失败。
tm.transaction.netty.delaytime = 5

#业务模块与TxManager之间通讯的心跳时间（单位：秒）
tm.transaction.netty.hearttime = 15

#存储到redis下的数据最大保存时间（单位：秒）
#该字段仅代表的事务模块数据的最大保存时间，补偿数据会永久保存。
tm.redis.savemaxtime=30

#socket server Socket对外服务端口
#TxManager的LCN协议的端口
tm.socket.port=9999

#最大socket连接数
#TxManager最大允许的建立连接数量
tm.socket.maxconnection=100

#事务自动补偿 (true:开启，false:关闭)
# 说明：
# 开启自动补偿以后，必须要配置 tm.compensate.notifyUrl 地址，仅当tm.compensate.notifyUrl 在请求补偿确认时返回success或者SUCCESS时，才会执行自动补偿，否则不会自动补偿。
# 关闭自动补偿，当出现数据时也会 tm.compensate.notifyUrl 地址。
# 当tm.compensate.notifyUrl 无效时，不影响TxManager运行，仅会影响自动补偿。
tm.compensate.auto=false

#事务补偿记录回调地址(rest api 地址，post json格式)
#请求补偿是在开启自动补偿时才会请求的地址。请求分为两种：1.补偿决策，2.补偿结果通知，可通过通过action参数区分compensate为补偿请求、notify为补偿通知。
#*注意当请求补偿决策时，需要补偿服务返回"SUCCESS"字符串以后才可以执行自动补偿。
#请求补偿结果通知则只需要接受通知即可。
#请求补偿的样例数据格式:
#{"groupId":"TtQxTwJP","action":"compensate","json":"{"address":"133.133.5.100:8081","className":"com.example.demo.service.impl.DemoServiceImpl","currentTime":1511356150413,"data":"C5IBLWNvbS5leGFtcGxlLmRlbW8uc2VydmljZS5pbXBsLkRlbW9TZXJ2aWNlSW1wbAwSBHNhdmUbehBqYXZhLmxhbmcuT2JqZWN0GAAQARwjeg9qYXZhLmxhbmcuQ2xhc3MYABABJCo/cHVibGljIGludCBjb20uZXhhbXBsZS5kZW1vLnNlcnZpY2UuaW1wbC5EZW1vU2VydmljZUltcGwuc2F2ZSgp","groupId":"TtQxTwJP","methodStr":"public int com.example.demo.service.impl.DemoServiceImpl.save()","model":"demo1","state":0,"time":36,"txGroup":{"groupId":"TtQxTwJP","hasOver":1,"isCompensate":0,"list":[{"address":"133.133.5.100:8899","isCompensate":0,"isGroup":0,"kid":"wnlEJoSl","methodStr":"public int com.example.demo.service.impl.DemoServiceImpl.save()","model":"demo2","modelIpAddress":"133.133.5.100:8082","channelAddress":"/133.133.5.100:64153","notify":1,"uniqueKey":"bc13881a5d2ab2ace89ae5d34d608447"}],"nowTime":0,"startTime":1511356150379,"state":1},"uniqueKey":"be6eea31e382f1f0878d07cef319e4d7"}"}
#请求补偿的返回数据样例数据格式:
#SUCCESS
#请求补偿结果通知的样例数据格式:
#{"resState":true,"groupId":"TtQxTwJP","action":"notify"}
tm.compensate.notifyUrl=http://ip:port/path

#补偿失败，再次尝试间隔（秒），最大尝试次数3次，当超过3次即为补偿失败,失败的数据依旧还会存在TxManager下。
tm.compensate.tryTime=30

#各事务模块自动补偿的时间上限(毫秒)
#指的是模块执行自动超时的最大时间，该最大时间若过段会导致事务机制异常，该时间必须要模块之间通讯的最大超过时间。
#例如，若模块A与模块B，请求超时的最大时间是5秒，则建议改时间至少大于5秒。
tm.compensate.maxWaitTime=5000
#######################################LCN-end#################################################




logging.level.com.codingapi=debug

2、服务中加入分布式事务管理（加入LCN事务管理）

pom.xml中添加LCN框架的依赖

  1 <?xml version="1.0" encoding="UTF-8"?>
  2 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  4     <modelVersion>4.0.0</modelVersion>
  5 
  6     <groupId>com.prise</groupId>
  7     <artifactId>springcloud-lcn-demo</artifactId>
  8     <version>4.1.0</version>
  9     <packaging>pom</packaging>
 10 
 11     <name>springcloud-lcn-demo</name>
 12 
 13     <modules>
 14         <module>mybatis-demo/springcloud-mybatis-demo1</module>
 15         <module>mybatis-demo/springcloud-mybatis-demo2</module>
 16         <module>mybatis-demo/springcloud-mybatis-demo3</module>
 17     </modules>
 18 
 19 
 20     <parent>
 21         <groupId>org.springframework.boot</groupId>
 22         <artifactId>spring-boot-starter-parent</artifactId>
 23         <version>1.5.9.RELEASE</version>
 24         <relativePath/> <!-- lookup parent from repository -->
 25     </parent>
 26 
 27     <properties>
 28         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 29         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 30 
 31         <java.version>1.8</java.version>
 32         <maven.compile.source>1.7</maven.compile.source>
 33         <maven.compile.target>1.7</maven.compile.target>
 34         <spring-cloud.version>Edgware.RELEASE</spring-cloud.version>
 35         <!--lcn的版本-->
 36         <lcn.last.version>4.1.0</lcn.last.version>
 37 
 38     </properties>
 39 
 40     <dependencies>
 41         <!--LCN基于springcloud的分布式事务框架-->
 42         <dependency>
 43             <groupId>com.codingapi</groupId>
 44             <artifactId>transaction-springcloud</artifactId>
 45             <version>${lcn.last.version}</version>
 46             <exclusions>
 47                 <exclusion>
 48                     <groupId>org.slf4j</groupId>
 49                     <artifactId>*</artifactId>
 50                 </exclusion>
 51             </exclusions>
 52         </dependency>
 53         <!--LCN基于关系型数据模块的封装-->
 54         
 55         <dependency>
 56             <groupId>com.codingapi</groupId>
 57             <artifactId>tx-plugins-db</artifactId>
 58             <version>${lcn.last.version}</version>
 59             <exclusions>
 60                 <exclusion>
 61                     <groupId>org.slf4j</groupId>
 62                     <artifactId>*</artifactId>
 63                 </exclusion>
 64             </exclusions>
 65         </dependency>
 66         <!--spring-cloud注册中心-->
 67         <dependency>
 68             <groupId>org.springframework.cloud</groupId>
 69             <artifactId>spring-cloud-starter-eureka</artifactId>
 70         </dependency>
 71         <dependency>
 72             <groupId>org.springframework.cloud</groupId>
 73             <artifactId>spring-cloud-starter-feign</artifactId>
 74         </dependency>
 75         <dependency>
 76             <groupId>org.springframework.boot</groupId>
 77             <artifactId>spring-boot-starter-test</artifactId>
 78             <scope>test</scope>
 79         </dependency>
 80 
 81     </dependencies>
 82 
 83     <dependencyManagement>
 84         <dependencies>
 85             <dependency>
 86                 <groupId>org.springframework.cloud</groupId>
 87                 <artifactId>spring-cloud-dependencies</artifactId>
 88                 <version>${spring-cloud.version}</version>
 89                 <type>pom</type>
 90                 <scope>import</scope>
 91             </dependency>
 92         </dependencies>
 93     </dependencyManagement>
 94 
 95     <build>
 96         <plugins>
 97             <plugin>
 98                 <groupId>org.apache.maven.plugins</groupId>
 99                 <artifactId>maven-compiler-plugin</artifactId>
100                 <configuration>
101                     <source>${maven.compile.source}</source>
102                     <target>${maven.compile.target}</target>
103                     <encoding>${project.build.sourceEncoding}</encoding>
104                 </configuration>
105             </plugin>
106 
107             <plugin>
108                 <groupId>org.springframework.boot</groupId>
109                 <artifactId>spring-boot-maven-plugin</artifactId>
110             </plugin>
111         </plugins>
112     </build>
113 
114 
115 </project>

transaction-springcloud LCN springcloud rpc框架扩展支持，我们这里使用的是springcloud
tx-plugins-db 是LCN 对关系型数据库的插件支持，我们这里使用的是mysql数据库.我是将三个微服务放在一个包里一起的，但是每个微服务都独立一个端口和数据库。

配置TXmanager的访问地址和端口号

 1 #feign.hystrix.enabled=true
 2 
 3 spring.datasource.driver-class-name = com.mysql.jdbc.Driver
 4 spring.datasource.url= jdbc:mysql://localhost:3306/database0
 5 spring.datasource.username= root
 6 spring.datasource.password= 1234
 7 spring.datasource.initialize =  true
 8 init-db= true
 9 
10 spring.application.name = demo1
11 server.port = 8085
12 #${random.int[9000,9999]}
13 eureka.client.service-url.defaultZone=http://localhost:8083/eureka/
14 
15 #txmanagerå?°å??
16 tm.manager.url=http://127.0.0.1:8899/tx/manager/
17 
18 logging.level.com.codingapi=debug
19 
20 spring.jpa.show-sql=true

分布式事务发起方
需要实现TxManagerTxUrlService和TxManagerHttpRequestService这两个接口，并作为bean注入到spring中。处理http请求和对服务器的连接。

　　　　　实现 TxManagerTxUrlService

 1 @Service
 2 public class TxManagerTxUrlServiceImpl implements TxManagerTxUrlService{
 3 
 4 
 5     @Value("${tm.manager.url}")
 6     private String url;
 7 
 8     @Override
 9     public String getTxUrl() {
10         System.out.println("load tm.manager.url ");
11         return url;
12     }
13 }

这个方法是为了获得连接LCN事务管理器的。

实现 TxManagerHttpRequestService

 1 @Service
 2 public class TxManagerHttpRequestServiceImpl implements TxManagerHttpRequestService{
 3 
 4     @Override
 5     public String httpGet(String url) {
 6         System.out.println("httpGet-start");
 7         String res = HttpUtils.get(url);
 8         System.out.println("httpGet-end");
 9         return res;
10     }
11 
12     @Override
13     public String httpPost(String url, String params) {
14         System.out.println("httpPost-start");
15         String res = HttpUtils.post(url,params);
16         System.out.println("httpPost-end");
17         return res;
18     }
19 }

这个文件的作用是：作为事务的发起者，要开启一个事务组，要主动连接上分布式事务管理框架。

加入分布式事务注解@TxTransaction(isStart = true)，开启分布式事务。isStart = true声明为分布式事务发起方

 1 @Service
 2 public class DemoServiceImpl implements DemoService {
 3 
 4     @Autowired
 5     private Demo2Feign demo2Feign;
 6     @Autowired
 7     private Demo3Feign demo3Feign;
 8 
 9     @Autowired
10     private UserMapper userMapper;
11 
12     private static final Logger logger = LoggerFactory.getLogger(DemoServiceImpl.class);
13 
14     @Override
15     public List<User> list() {
16         return userMapper.findAll();
17     }
18 
19     @Override
20     @TxTransaction(isStart = true)
21     @Transactional
22     public int save() throws Exception {
23 
24         int rs1 = userMapper.save("zhangsan", "shanghai");
25         logger.info("本地服务数据插入成功");
26          int rs2 = demo2Feign.save();
27 //        Integer rs2 = restTemplate.getForObject("http://127.0.0.1:8087/demo/save", Integer.class);
28 //        System.out.println(rs2); 
29         logger.info("远程服务2号数据插入成功"+rs2);
30 //        if(0 ==rs2) {
31 //            throw new RuntimeException("远程访问出错，全部回滚");
32 //        }
33          int rs3 = demo3Feign.save();
34          logger.info("远程服务3号数据插入成功");
35          logger.info("插入" + (rs1 + rs2 + rs3) + "条记录");
36          logger.info("制造异常");
37          int v = 100 / 0;
38 //        if(1==rs2 ) {
39 //            throw new RuntimeException("本地出错，全部回滚");
40 //        }
41         return rs1 + rs2;
42     }
43 }

如上代码执行完成以后所有参与此分布式事务模块都将回滚事务。

分布式事务被调用方

需要实现TxManagerTxUrlService个接口，并作为bean注入到spring中。处理对服务器的连接。

 1 @Service
 2 public class TxManagerTxUrlServiceImpl implements TxManagerTxUrlService{
 3 
 4 
 5     @Value("${tm.manager.url}")
 6     private String url;
 7 
 8     @Override
 9     public String getTxUrl() {
10         System.out.println("load tm.manager.url ");
11         return url;
12     }
13 }

加入分布式事务注解@TxTransaction，或者实现ITxTransaction接口，开启分布式事务。等待起调方发起事务

 1 @Service
 2 public class DemoServiceImpl implements DemoService,ITxTransaction{
 3 
 4     @Autowired
 5     private PeopleMapper peopleMapper;
 6 
 7     private static final Logger logger = LoggerFactory.getLogger(DemoServiceImpl.class);
 8 
 9 
10     @Override
11     public List<People> list() {
12         return peopleMapper.findAll();
13     }
14 
15 
16     @Override
17     @Transactional
18     public int save() {
19         try {
20         
21         int rs = peopleMapper.save("男", "22");
22         logger.info("插入" + rs + "条记录");
23         int a = 1/1;
24         System.out.println(a);
25         return rs;
26         }catch (Exception e) {
27             logger.error(e.getMessage());
28             return 0;
29         }
30     }
31 }

说明：在使用LCN分布式事务时，只需要将事务的开始方法添加@TxTransaction(isStart=true)注解即可,在参与方添加@TxTransaction或者实现ITxTransaction接口即可。

代码下载地址：https://gitee.com/fengyuduke/my_open_resources/blob/master/demo.zip

以上是关于LCN分布式事务管理的主要内容，如果未能解决你的问题，请参考以下文章

spring boot 分布式事务解决方案LCN

TX-LCN：分布式事务框架