LCN分布式事务管理（一）

2023-08-01 14:20:10

前言

好久没写东西了，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框架的依赖

 <?xml version="1.0" encoding="UTF-8"?>

 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

     <modelVersion>4.0.0</modelVersion>

     <groupId>com.prise</groupId>

     <artifactId>springcloud-lcn-demo</artifactId>

     <version>4.1.0</version>

     <packaging>pom</packaging>

     <name>springcloud-lcn-demo</name>

     <modules>

         <module>mybatis-demo/springcloud-mybatis-demo1</module>

         <module>mybatis-demo/springcloud-mybatis-demo2</module>

         <module>mybatis-demo/springcloud-mybatis-demo3</module>

     </modules>

     <parent>

         <groupId>org.springframework.boot</groupId>

         <artifactId>spring-boot-starter-parent</artifactId>

         <version>1.5.9.RELEASE</version>

         <relativePath/> <!-- lookup parent from repository -->

     </parent>

     <properties>

         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

         <java.version>1.8</java.version>

         <maven.compile.source>1.7</maven.compile.source>

         <maven.compile.target>1.7</maven.compile.target>

         <spring-cloud.version>Edgware.RELEASE</spring-cloud.version>

         <!--lcn的版本-->

         <lcn.last.version>4.1.0</lcn.last.version>

     </properties>

     <dependencies>

         <!--LCN基于springcloud的分布式事务框架-->

         <dependency>

             <groupId>com.codingapi</groupId>

             <artifactId>transaction-springcloud</artifactId>

             <version>${lcn.last.version}</version>

             <exclusions>

                 <exclusion>

                     <groupId>org.slf4j</groupId>

                     <artifactId>*</artifactId>

                 </exclusion>

             </exclusions>

         </dependency>

         <!--LCN基于关系型数据模块的封装-->

         <dependency>

             <groupId>com.codingapi</groupId>

             <artifactId>tx-plugins-db</artifactId>

             <version>${lcn.last.version}</version>

             <exclusions>

                 <exclusion>

                     <groupId>org.slf4j</groupId>

                     <artifactId>*</artifactId>

                 </exclusion>

             </exclusions>

         </dependency>

         <!--spring-cloud注册中心-->

         <dependency>

             <groupId>org.springframework.cloud</groupId>

             <artifactId>spring-cloud-starter-eureka</artifactId>

         </dependency>

         <dependency>

             <groupId>org.springframework.cloud</groupId>

             <artifactId>spring-cloud-starter-feign</artifactId>

         </dependency>

         <dependency>

             <groupId>org.springframework.boot</groupId>

             <artifactId>spring-boot-starter-test</artifactId>

             <scope>test</scope>

         </dependency>

     </dependencies>

     <dependencyManagement>

         <dependencies>

             <dependency>

                 <groupId>org.springframework.cloud</groupId>

                 <artifactId>spring-cloud-dependencies</artifactId>

                 <version>${spring-cloud.version}</version>

                 <type>pom</type>

                 <scope>import</scope>

             </dependency>

         </dependencies>

     </dependencyManagement>

     <build>

         <plugins>

             <plugin>

                 <groupId>org.apache.maven.plugins</groupId>

                 <artifactId>maven-compiler-plugin</artifactId>

                 <configuration>

                     <source>${maven.compile.source}</source>

                     <target>${maven.compile.target}</target>

                     <encoding>${project.build.sourceEncoding}</encoding>

                 </configuration>

             </plugin>

             <plugin>

                 <groupId>org.springframework.boot</groupId>

                 <artifactId>spring-boot-maven-plugin</artifactId>

             </plugin>

         </plugins>

     </build>

 </project>

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

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

 #feign.hystrix.enabled=true

 spring.datasource.driver-class-name = com.mysql.jdbc.Driver

 spring.datasource.url= jdbc:mysql://localhost:3306/database0

 spring.datasource.username= root

 spring.datasource.password= 1234

 spring.datasource.initialize =  true

 init-db= true

 spring.application.name = demo1

 server.port = 8085

 #${random.int[9000,9999]}

 eureka.client.service-url.defaultZone=http://localhost:8083/eureka/

 #txmanagerå°å

 tm.manager.url=http://127.0.0.1:8899/tx/manager/

 logging.level.com.codingapi=debug

 spring.jpa.show-sql=true

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

　　　　　实现 TxManagerTxUrlService

 @Service

 public class TxManagerTxUrlServiceImpl implements TxManagerTxUrlService{

     @Value("${tm.manager.url}")

     private String url;

     @Override

     public String getTxUrl() {

         System.out.println("load tm.manager.url ");

         return url;

     }

 }

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

实现 TxManagerHttpRequestService

 @Service

 public class TxManagerHttpRequestServiceImpl implements TxManagerHttpRequestService{

     @Override

     public String httpGet(String url) {

         System.out.println("httpGet-start");

         String res = HttpUtils.get(url);

         System.out.println("httpGet-end");

         return res;

     }

     @Override

     public String httpPost(String url, String params) {

         System.out.println("httpPost-start");

         String res = HttpUtils.post(url,params);

         System.out.println("httpPost-end");

         return res;

     }

 }

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

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

 @Service

 public class DemoServiceImpl implements DemoService {

     @Autowired

     private Demo2Feign demo2Feign;

     @Autowired

     private Demo3Feign demo3Feign;

     @Autowired

     private UserMapper userMapper;

     private static final Logger logger = LoggerFactory.getLogger(DemoServiceImpl.class);

     @Override

     public List<User> list() {

         return userMapper.findAll();

     }

     @Override

     @TxTransaction(isStart = true)

     @Transactional

     public int save() throws Exception {

         int rs1 = userMapper.save("zhangsan", "shanghai");

         logger.info("本地服务数据插入成功");

          int rs2 = demo2Feign.save();

 //        Integer rs2 = restTemplate.getForObject("http://127.0.0.1:8087/demo/save", Integer.class);

 //        System.out.println(rs2);

         logger.info("远程服务2号数据插入成功"+rs2);

 //        if(0 ==rs2) {

 //            throw new RuntimeException("远程访问出错，全部回滚");

 //        }

          int rs3 = demo3Feign.save();

          logger.info("远程服务3号数据插入成功");

          logger.info("插入" + (rs1 + rs2 + rs3) + "条记录");

          logger.info("制造异常");

          int v = 100 / 0;

 //        if(1==rs2 ) {

 //            throw new RuntimeException("本地出错，全部回滚");

 //        }

         return rs1 + rs2;

     }

 }

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

分布式事务被调用方

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

 @Service

 public class TxManagerTxUrlServiceImpl implements TxManagerTxUrlService{

     @Value("${tm.manager.url}")

     private String url;

     @Override

     public String getTxUrl() {

         System.out.println("load tm.manager.url ");

         return url;

     }

 }

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

 @Service

 public class DemoServiceImpl implements DemoService,ITxTransaction{

     @Autowired

     private PeopleMapper peopleMapper;

     private static final Logger logger = LoggerFactory.getLogger(DemoServiceImpl.class);

     @Override

     public List<People> list() {

         return peopleMapper.findAll();

     }

     @Override

     @Transactional

     public int save() {

         try {

         int rs = peopleMapper.save("男", "22");

         logger.info("插入" + rs + "条记录");

         int a = 1/1;

         System.out.println(a);

         return rs;

         }catch (Exception e) {

             logger.error(e.getMessage());

             return 0;

         }

     }

 }

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

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

码农公寓

前言

框架介绍

核心步骤

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

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

相关文章