基于springboot 2.2.2+sharding+seata实现分布式事务

一、背景
　　1.1 分布式事务产生
　　在微服务环境下，我们会根据不同的业务拆分成不同的服务，比如采购订单服务、采购送货单服务、供应商管理服务等。每个服务都有自己独立的数据库并且是独立运行的，互不影响。服务与服务之间通讯采用RPC远程调用技术。每个服务中都有自己独立的数据源，也就是自己独立的本地事务。两个服务之间相互通讯的时候，两个本地事务互不影响，从而就出现了分布式事务。
　　1.2 模式选择
　　分布式事务框架有很多，本解决方案基于seata框架的AT模式实现分布式事务控制，同时seata也支持tcc和saga模式，由于这两种模式在代码量和技术难度上都要求较高，在选择时可以根据实际业务场景需求进行，在此也提供柔性事务选择取舍特征。有关分布式事务的其他模式，在此不做详细赘述。
柔性事务选择取舍特征：

使用AT模式有两个前提需要提前注意的：
　　　　1.基于支持本地ACID事务的关系型数据库。
　　　　2.Java应用，通过JDBC访问数据库。
　　1.3 系统架构
　　Seata AT过程模型如下：
　
　　Sharding和Seata集成架构图：
　　
　　1.4 业务场景
　　在采购业务领域中有一项业务是创建送货单，在创建送货单时要进行业务逻辑判断，是不是VMI的送货单，如果是，此时需要创建采购计划。在创建采购计划时也需要判断计划的来源是不是VMI，如果是，需要创建采购需求。业务流程图如下：

以上业务场景对真实业务场景做了脱敏。
二、java代码
　　2.1 技术栈
　　代码技术栈使用：spring boot 2.2.2、mybatis2.1.3、druid1.1.10、sharding4.1.0、mysql5.7、idgenerator1.4.2、eureka2.2.1、openfeign2.2.1、seata1.2、分布式事务服务端使用seata1.3。
　　2.2 采购送货单服务
　　2.2.1 工程结构

　　2.2.2 pom文件依赖包

<dependencies><!-- spring boot包引入 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope><exclusions><exclusion><groupId>org.junit.vintage</groupId><artifactId>junit-vintage-engine</artifactId></exclusion></exclusions></dependency><!-- mybatisplus依赖包引入，引入MyBatis-Plus之后请不要再次引入 MyBatis 以及 MyBatis-Spring，以避免因版本差异导致的问题。 --><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.1.2</version></dependency><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus</artifactId><version>3.1.2</version></dependency><!-- mybatis分页插件 --><dependency><groupId>com.github.pagehelper</groupId><artifactId>pagehelper-spring-boot-starter</artifactId><version>1.2.12</version></dependency><!-- mysql驱动包依赖包引入 --><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><scope>runtime</scope></dependency><!-- druid连接池依赖包引入 --><dependency><groupId>com.alibaba</groupId><artifactId>druid-spring-boot-starter</artifactId><version>1.1.10</version></dependency><!-- sharding-jdbc数据库读写分离依赖包引入 --><dependency><groupId>org.apache.shardingsphere</groupId><artifactId>sharding-jdbc-core</artifactId><version>4.1.0</version></dependency><dependency><groupId>org.apache.shardingsphere</groupId><artifactId>sharding-jdbc-spring-boot-starter</artifactId><version>4.1.0</version></dependency><!-- 使用sharding的BASE事务依赖包引入 --><dependency><groupId>org.apache.shardingsphere</groupId><artifactId>sharding-transaction-base-seata-at</artifactId><version>4.1.0</version></dependency><!-- sharding学花算法实现的全局分布式ID依赖包引入 --><dependency><groupId>com.dangdang</groupId><artifactId>sharding-jdbc-self-id-generator</artifactId><version>1.4.2</version></dependency><!-- seata分布式事务依赖引入 --><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-alibaba-seata</artifactId><version>2.2.0.RELEASE</version><exclusions><exclusion><artifactId>seata-all</artifactId><groupId>io.seata</groupId></exclusion></exclusions></dependency><!-- 不使用stater中依赖的seata，使用1.2.0版本 --><dependency><groupId>io.seata</groupId><artifactId>seata-all</artifactId><version>1.2.0</version></dependency><dependency><groupId>io.seata</groupId><artifactId>seata-spring-boot-starter</artifactId><version>1.2.0</version></dependency><!--eureka客户端依赖包引入 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId><version>2.2.1.RELEASE</version></dependency><!--openfeign依赖包引入 --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-openfeign</artifactId><version>2.2.1.RELEASE</version></dependency>			</dependencies>

2.2.3 application.properties文件配置

#服务端口
server.port=8081
#tomcat编码
server.tomcat.uri-encoding=UTF-8
spring.aop.proxy-target-class=true
spring.application.name=delivery-note
spring.cloud.alibaba.seata.tx-service-group=delivery-note-group
spring.main.allow-bean-definition-overriding=true
#表示是否将自己注册进EurekaServer默认为true
eureka.client.fetch-registry=true
#是否从EurekaServer抓取已有的注册信息，默认为true。单节点无所谓，集群必须设置为true才能配合ribbon使用负载均衡
eureka.client.register-with-eureka=true
#eureka服务器地址单机
eureka.client.service-url.defaultZone=http://127.0.0.1:8181/eureka/
#注册到eureka中的服务名称
eureka.instance.instance-id=delivery-note
#访问路径可以显示IP地址
eureka.instance.prefer-ip-address=true
#feign客户端读取超时时间(单位：毫秒)
feign.client.config.default.read-timeout=15000#feign客户端连接超时时间(单位：毫秒)
feign.client.config.default.connect-timeout=15000
#当OkToRetryOnAllOperations设置为false时，只会对get请求进行重试。如果设置为true，便会对所有的请求进行重试，如果是put或post等写操作，如果服务器接口没做幂等性，会产生不好的结果，所以OkToRetryOnAllOperations慎用。
ribbon.OkToRetryOnAllOperations=false
#数据源名称，多数据源以逗号分隔
spring.shardingsphere.datasource.names=master,slave0,slave1
#主数据源
#数据库连接池类名称
spring.shardingsphere.datasource.master.type=com.alibaba.druid.pool.DruidDataSource
#数据库驱动类名
spring.shardingsphere.datasource.master.driver-class-name=com.mysql.cj.jdbc.Driver
#数据库URL连接
spring.shardingsphere.datasource.master.url=jdbc:mysql://192.168.145.183:3306/order?useUnicode=true&characterEncoding=utf8&tinyInt1isBit=false&useSSL=false&serverTimezone=GMT
#数据库用户名
spring.shardingsphere.datasource.master.username=root
#数据库密码
spring.shardingsphere.datasource.master.password=123456
#初始化大小，最小，最大
spring.shardingsphere.datasource.master.initial-size=5
spring.shardingsphere.datasource.master.min-idle=5
spring.shardingsphere.datasource.master.max-active=20
#配置获取连接等待超时的时间
spring.shardingsphere.datasource.master.max-wait=60000
#从数据源
spring.shardingsphere.datasource.slave0.type=com.alibaba.druid.pool.DruidDataSource
spring.shardingsphere.datasource.slave0.driver-class-name=com.mysql.cj.jdbc.Driver
spring.shardingsphere.datasource.slave0.url=jdbc:mysql://192.168.145.184:3306/order?useUnicode=true&characterEncoding=utf8&tinyInt1isBit=false&useSSL=false&serverTimezone=GMT
spring.shardingsphere.datasource.slave0.username=root
spring.shardingsphere.datasource.slave0.password=123456
#初始化大小，最小，最大
spring.shardingsphere.datasource.slave0.initial-size=5
spring.shardingsphere.datasource.slave0.min-idle=5
spring.shardingsphere.datasource.slave0.max-active=20
#配置获取连接等待超时的时间
spring.shardingsphere.datasource.slave0.max-wait=60000
#从数据源
spring.shardingsphere.datasource.slave1.type=com.alibaba.druid.pool.DruidDataSource
spring.shardingsphere.datasource.slave1.driver-class-name=com.mysql.cj.jdbc.Driver
spring.shardingsphere.datasource.slave1.url=jdbc:mysql://192.168.145.185:3306/order?useUnicode=true&characterEncoding=utf8&tinyInt1isBit=false&useSSL=false&serverTimezone=GMT
spring.shardingsphere.datasource.slave1.username=root
spring.shardingsphere.datasource.slave1.password=123456
#初始化大小，最小，最大
spring.shardingsphere.datasource.slave1.initial-size=5
spring.shardingsphere.datasource.slave1.min-idle=5
spring.shardingsphere.datasource.slave1.max-active=20
#配置获取连接等待超时的时间
spring.shardingsphere.datasource.slave1.max-wait=60000
#读写分离最终的数据源名称
spring.shardingsphere.masterslave.name=ms
#主库数据源名称
spring.shardingsphere.masterslave.master-data-source-name=master
#从库数据源名称，多个从库数据源用逗号分隔
spring.shardingsphere.masterslave.slave-data-source-names=slave0,slave1
#读写分离配置负载均衡算法名称
spring.shardingsphere.masterslave.load-balance-algorithm-type=round_robin
#开启SQL显示
spring.shardingsphere.props.sql.show=true
#mybatisplus配置mapper的xml位置
mybatis-plus.mapper-locations=classpath:mybatis/mapper/**/*.xml
#mybatisplus配置别名扫描
mybatis-plus.typeAliasesPackage=com.general.**.entity
#pagehelper分页插件
pagehelper.helperDialect=oracle
pagehelper.reasonable=true
pagehelper.supportMethodsArguments=true
pagehelper.params=count=countSql
logging.level.root=info

注：ribbon.OkToRetryOnAllOperations=false这句话一定要设置，否则post会出现重试，导致产生脏数据
　　2.2.4 seata配置文件
　　1.registry.conf

registry {
    # file 、nacos 、eureka、redis、zk、consul、etcd3、sofatype = "file"  file {
    name = "file.conf"}
}config {
    # file、nacos 、apollo、zk、consul、etcd3、springCloudConfigtype = "file"  file {
    name = "file.conf"}
}

2.file.conf

transport {
    type = "TCP"server = "NIO"heartbeat = trueenableClientBatchSendRequest = truethreadFactory {
    bossThreadPrefix = "NettyBoss"workerThreadPrefix = "NettyServerNIOWorker"serverExecutorThread-prefix = "NettyServerBizHandler"shareBossWorker = falseclientSelectorThreadPrefix = "NettyClientSelector"clientSelectorThreadSize = 1clientWorkerThreadPrefix = "NettyClientWorkerThread"bossThreadSize = 1workerThreadSize = "default"}shutdown {
    wait = 3}serialization = "seata"compressor = "none"
}
service {
    vgroupMapping.delivery-note-group = "seata-server"seata-server.grouplist = "192.168.145.185:8091"enableDegrade = falsedisableGlobalTransaction = false
}client {
    rm {
    asyncCommitBufferLimit = 10000lock {
    retryInterval = 10retryTimes = 30retryPolicyBranchRollbackOnConflict = true}reportRetryCount = 5tableMetaCheckEnable = falsereportSuccessEnable = false}tm {
    commitRetryCount = 5rollbackRetryCount = 5}undo {
    dataValidation = truelogSerialization = "jackson"logTable = "undo_log"}log {
    exceptionRate = 100}
}

注：vgroupMapping.delivery-note-group = “seata-server”
seata-server.grouplist = “192.168.145.185:8091”
这个配置一定要注意，默认格式是：
vgroupMapping.事务分组名称 = “事务分组名值”
事务分组名值.grouplist= “127.0.0.1:8091”
这里特别注意，如果seata服务器是其他IP地址，这里事务分组名值绝对不能写default，如果写default，seata会忽略下面的事务分组名值.grouplist= "127.0.0.1:8091"这句话
直接给事务分组名值.grouplist的值设置为127.0.0.1:8091，从而导致我们设置自己的seata服务器不生效。
　　3. seata.conf

client {
    application.id = delivery-notetransaction.service.group = delivery-note-group
}

注：事务交给seata管理，业务表必须有主键
　　2.2.5 启动类

@EnableEurekaClient
@EnableFeignClients(defaultConfiguration = FeignConfig.class)
@MapperScan("com.general.**.mapper")
@ComponentScan(value = "com.general")
@SpringBootApplication(exclude = {
     DruidDataSourceAutoConfigure.class, DataSourceAutoConfiguration.class })
public class DeliveryNoteApplication {
    public static void main(String[] args) {
    SpringApplication.run(DeliveryNoteApplication.class, args);}/*** 创建分布式ID对象* @return 分布式ID对象*/@Beanpublic IdGenerator getIdGenerator() {
    return new CommonSelfIdGenerator();}
}

2.2.6 配置类
　　1.Feign的配置类

@Configuration
public class FeignConfig {
    @BeanLogger.Level feignLoggerLevel() {
    //设置feign的日志级别为：记录请求和响应的头文件，正文和元数据。return Logger.Level.FULL;}
}

2.MybatisPlus配置类

@EnableTransactionManagement
@Configuration
public class MybatisPlusConfig {
    /*** 分页插件* @return 分页插件对象*/@Beanpublic PaginationInterceptor paginationInterceptor() {
    return new PaginationInterceptor();}
}

2.2.7 业务请求控制

@RestController
@RequestMapping("/deliverynoteweb")
public class DeliveryNoteWeb {
    @Autowiredprivate DeliveryNoteService deliveryNoteService;@Autowiredprivate IdGenerator idGenerator;/*** 查询所有采购送货单* @return 所有采购送货单*/@RequestMapping("/query")public Object query() {
    Page<DeliveryNote> page = new Page<DeliveryNote>(100, 50);return deliveryNoteService.findByAll(page);}/*** 保存采购送货单-正常* @return 执行信息*/@RequestMapping("/insertnormal")public Object insertNormal() {
    DeliveryNote deliveryNote = new DeliveryNote();//这里使用sharding的一个分布式ID组件，底层是使用snowflake算法实现。deliveryNote.setId(idGenerator.generateId().longValue());deliveryNote.setCode("测试采购送货单编码" + deliveryNote.getId());deliveryNote.setName("测试采购送货单名称" + deliveryNote.getId());deliveryNote.setNum(10D);return deliveryNoteService.insertNormal(deliveryNote);}/*** 保存采购送货单-异常* @return 执行信息*/@RequestMapping("/insertexception")public Object insertException() {
    DeliveryNote deliveryNote = new DeliveryNote();//这里使用sharding的一个分布式ID组件，底层是使用snowflake算法实现。deliveryNote.setId(idGenerator.generateId().longValue());deliveryNote.setCode("测试采购送货单编码" + deliveryNote.getId());deliveryNote.setName("测试采购送货单名称" + deliveryNote.getId());deliveryNote.setNum(10D);return deliveryNoteService.insertException(deliveryNote);}
}

2.2.8 业务接口

public interface DeliveryNoteService {
    /*** 查询所有采购送货单* @return 采购送货单集合*/public Page<DeliveryNote> findByAll(Page<DeliveryNote> page);/*** 保存采购送货单正常* @param deliveryNote 采购送货单* @return 执行信息*/public String insertNormal(DeliveryNote deliveryNote);/*** 保存采购送货单异常* @param deliveryNote 采购送货单* @return 执行信息*/public String insertException(DeliveryNote deliveryNote);/*** 更新采购送货单* @param deliveryNote 采购送货单*/public void update(DeliveryNote deliveryNote);/*** 删除采购送货单* @param deliveryNote 采购送货单*/public void delete(DeliveryNote deliveryNote);}**2.2.9 业务接口实现**
@Service
public class DeliveryNoteServiceImpl implements DeliveryNoteService {
    @Autowiredprivate DeliveryNoteMapper deliveryNoteMapper;@Autowiredprivate PurchasePlanService purchasePlanService;/*** 查询所有采购送货单* @return 采购送货单集合*/@Overridepublic Page<DeliveryNote> findByAll(Page<DeliveryNote> page) {
    return page.setRecords(deliveryNoteMapper.findByAllPage(page));}/*** 保存采购送货单* @param deliveryNote*/@GlobalTransactional(rollbackFor = Exception.class)@Overridepublic String insertNormal(DeliveryNote deliveryNote) {
    deliveryNoteMapper.insert(deliveryNote);purchasePlanService.insertNormal();return deliveryNote.getId() + "送货单保存成功";}/*** 保存采购送货单异常* @param deliveryNote*/@GlobalTransactional(rollbackFor = Exception.class)@Overridepublic String insertException(DeliveryNote deliveryNote) {
    deliveryNoteMapper.insert(deliveryNote);purchasePlanService.insertException();throw new RuntimeException(deliveryNote.getId() + "送货单保存异常");}/*** 更新采购送货单* @param deliveryNote*/@GlobalTransactional(rollbackFor = Exception.class)@Overridepublic void update(DeliveryNote deliveryNote) {
    deliveryNoteMapper.update(deliveryNote);}/*** 删除采购送货单* @param deliveryNote*/@GlobalTransactional(rollbackFor = Exception.class)@Overridepublic void delete(DeliveryNote deliveryNote) {
    deliveryNoteMapper.delete(deliveryNote);}
}

注：@GlobalTransactional(rollbackFor = Exception.class)这个是全局事务控制注解，报任何异常，都会进行事务回滚，对于敏捷开发非常合适。如果需要做分布式事务控制添加此注解即可。
　　2.2.10 DAO接口

@Repository
public interface DeliveryNoteMapper extends BaseMapper<DeliveryNote> {
    /*** 查询所有采购送货单* @return*/public List<DeliveryNote> findByAllPage(Page<?> page);/*** 保存采购送货单* @param deliveryNote 采购送货单* @return */@Overridepublic int insert(DeliveryNote deliveryNote);/*** 更新采购送货单* @param deliveryNote 采购送货单*/public void update(DeliveryNote deliveryNote);/*** 删除采购送货单* @param deliveryNote 采购送货单*/public void delete(DeliveryNote deliveryNote);
}

2.2.11 实体类

public class DeliveryNote implements Serializable {
    /****/private static final long serialVersionUID = 1L;private Long id; //主键private String code; //送货单编码private String name; //送货单名称private String type; //送货单类型private Double num; //送货量public Long getId() {
    return id;}public void setId(Long id) {
    this.id = id;}public String getCode() {
    return code;}public void setCode(String code) {
    this.code = code;}public String getName() {
    return name;}public void setName(String name) {
    this.name = name;}public String getType() {
    return type;}public void setType(String type) {
    this.type = type;}public Double getNum() {
    return num;}public void setNum(Double num) {
    this.num = num;}
}

2.2.12 DeliveryNoteMapper.xml

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
<mapper namespace="com.general.deliverynote.mapper.DeliveryNoteMapper" ><resultMap id="BaseResultMap" type="com.general.deliverynote.entity.DeliveryNote"><id column="id" property="id" /><result column="code"  property="code" /><result column="name"  property="name" /><result column="type"  property="type" /><result column="num"  property="num" /></resultMap><select id="findByAllPage" resultMap="BaseResultMap">select * from delivery_note</select><insert id="insert" parameterType="com.general.deliverynote.entity.DeliveryNote">insert into delivery_note(id, code, name, type, num) values(#{
    id}, #{
    code}, #{
    name}, #{
    type}, #{
    num})</insert><update id="update" parameterType="com.general.deliverynote.entity.DeliveryNote">update delivery_note set code = #{
    code}, name = #{
    name}, type = #{
    type}, num = #{
    num}where id = #{
    id};</update><delete id="delete" parameterType="com.general.deliverynote.entity.DeliveryNote">delete from delivery_note where id = #{
    id};</delete> 
</mapper>

2.2.13 采购计划feign调用接口

@Component
@FeignClient(value = "PURCHASE-PLAN")
public interface PurchasePlanService {
    /*** 保存采购计划正常* @return 处理信息*/@GetMapping(value = "/purchaseplanweb/insertnormal")public String insertNormal();/*** 保存采购计划异常* @return 处理信息*/@GetMapping(value = "/purchaseplanweb/insertexception")public String insertException();
}

2.3 采购计划
　　2.3.1 工程结构

2.3.2 pom文件依赖包
　　　　和采购送货单服务相同，在此不贴了。
　　2.3.3 application.properties文件配置

#服务端口
server.port=8083
#tomcat编码
server.tomcat.uri-encoding=UTF-8
spring.aop.proxy-target-class=true
spring.application.name=purchase-plan
spring.cloud.alibaba.seata.tx-service-group=purchase-plan-group
spring.main.allow-bean-definition-overriding=true
#表示是否将自己注册进EurekaServer默认为true
eureka.client.fetch-registry=true
#是否从EurekaServer抓取已有的注册信息，默认为true。单节点无所谓，集群必须设置为true才能配合ribbon使用负载均衡
eureka.client.register-with-eureka=true
#eureka服务器地址单机
eureka.client.service-url.defaultZone=http://127.0.0.1:8181/eureka/
#注册到eureka中的服务名称
eureka.instance.instance-id=purchase-plan
#访问路径可以显示IP地址
eureka.instance.prefer-ip-address=true
#feign客户端读取超时时间(单位：毫秒)
feign.client.config.default.read-timeout=15000
#feign客户端连接超时时间(单位：毫秒)
feign.client.config.default.connect-timeout=15000
#当OkToRetryOnAllOperations设置为false时，只会对get请求进行重试。如果设置为true，便会对所有的请求进行重试，如果是put或post等写操作，如果服务器接口没做幂等性，会产生不好的结果，所以OkToRetryOnAllOperations慎用。
ribbon.OkToRetryOnAllOperations=false
#数据源名称，多数据源以逗号分隔
spring.shardingsphere.datasource.names=master,slave0,slave1
#数据库连接池类名称
spring.shardingsphere.datasource.master.type=com.alibaba.druid.pool.DruidDataSource
#数据库驱动类名
spring.shardingsphere.datasource.master.driver-class-name=com.mysql.cj.jdbc.Driver
#数据库URL连接
spring.shardingsphere.datasource.master.url=jdbc:mysql://192.168.145.183:3306/order?useUnicode=true&characterEncoding=utf8&tinyInt1isBit=false&useSSL=false&serverTimezone=GMT
#数据库用户名
spring.shardingsphere.datasource.master.username=root
#数据库密码
spring.shardingsphere.datasource.master.password=123456
#初始化大小，最小，最大
spring.shardingsphere.datasource.master.initial-size=5
spring.shardingsphere.datasource.master.min-idle=5
spring.shardingsphere.datasource.master.max-active=20
#配置获取连接等待超时的时间
spring.shardingsphere.datasource.master.max-wait=60000
spring.shardingsphere.datasource.slave0.type=com.alibaba.druid.pool.DruidDataSource
spring.shardingsphere.datasource.slave0.driver-class-name=com.mysql.cj.jdbc.Driver
spring.shardingsphere.datasource.slave0.url=jdbc:mysql://192.168.145.184:3306/order?useUnicode=true&characterEncoding=utf8&tinyInt1isBit=false&useSSL=false&serverTimezone=GMT
spring.shardingsphere.datasource.slave0.username=root
spring.shardingsphere.datasource.slave0.password=123456
#初始化大小，最小，最大
spring.shardingsphere.datasource.slave0.initial-size=5
spring.shardingsphere.datasource.slave0.min-idle=5
spring.shardingsphere.datasource.slave0.max-active=20
#配置获取连接等待超时的时间
spring.shardingsphere.datasource.slave0.max-wait=60000
spring.shardingsphere.datasource.slave1.type=com.alibaba.druid.pool.DruidDataSource
spring.shardingsphere.datasource.slave1.driver-class-name=com.mysql.cj.jdbc.Driver
spring.shardingsphere.datasource.slave1.url=jdbc:mysql://192.168.145.185:3306/order?useUnicode=true&characterEncoding=utf8&tinyInt1isBit=false&useSSL=false&serverTimezone=GMT
spring.shardingsphere.datasource.slave1.username=root
spring.shardingsphere.datasource.slave1.password=123456
#初始化大小，最小，最大
spring.shardingsphere.datasource.slave1.initial-size=5
spring.shardingsphere.datasource.slave1.min-idle=5
spring.shardingsphere.datasource.slave1.max-active=20
#配置获取连接等待超时的时间
spring.shardingsphere.datasource.slave1.max-wait=60000
#读写分离数据源名称
spring.shardingsphere.masterslave.name=ms
#主数据源名称
spring.shardingsphere.masterslave.master-data-source-name=master
#从数据源名称，多个从数据源用逗号分隔
spring.shardingsphere.masterslave.slave-data-source-names=slave0,slave1
#负载均衡算法名称
spring.shardingsphere.masterslave.load-balance-algorithm-type=round_robin
#开启SQL显示
spring.shardingsphere.props.sql.show=true
#mybatisplus配置mapper的xml位置
mybatis-plus.mapper-locations=classpath:mybatis/mapper/**/*.xml
#mybatisplus配置别名扫描
mybatis-plus.typeAliasesPackage=com.general.**.entity
logging.level.root=info

2.3.4 seata配置文件
　　1.registry.conf

registry {
    # file 、nacos 、eureka、redis、zk、consul、etcd3、sofatype = "file"file {
    name = "file.conf"}
}config {
    # file、nacos 、apollo、zk、consul、etcd3、springCloudConfigtype = "file"file {
    name = "file.conf"}
}

2.file.conf

transport {
    type = "TCP"server = "NIO"heartbeat = trueenableClientBatchSendRequest = truethreadFactory {
    bossThreadPrefix = "NettyBoss"workerThreadPrefix = "NettyServerNIOWorker"serverExecutorThread-prefix = "NettyServerBizHandler"shareBossWorker = falseclientSelectorThreadPrefix = "NettyClientSelector"clientSelectorThreadSize = 1clientWorkerThreadPrefix = "NettyClientWorkerThread"bossThreadSize = 1workerThreadSize = "default"}shutdown {
    wait = 3}serialization = "seata"compressor = "none"
}
service {
    vgroupMapping.purchase-plan-group = "seata-server"seata-server.grouplist = "192.168.145.185:8091"enableDegrade = falsedisableGlobalTransaction = false
}client {
    rm {
    asyncCommitBufferLimit = 10000lock {
    retryInterval = 10retryTimes = 30retryPolicyBranchRollbackOnConflict = true}reportRetryCount = 5tableMetaCheckEnable = falsereportSuccessEnable = false}tm {
    commitRetryCount = 5rollbackRetryCount = 5}undo {
    dataValidation = truelogSerialization = "jackson"logTable = "undo_log"}log {
    exceptionRate = 100}
}

3.seata.conf

client {
    application.id = purchase-plantransaction.service.group = purchase-plan-group
}

2.3.5 启动类

@EnableEurekaClient
@EnableFeignClients
@MapperScan("com.general.**.mapper")
@ComponentScan(value = "com.general")
@SpringBootApplication(exclude = {
     DruidDataSourceAutoConfigure.class, DataSourceAutoConfiguration.class })
public class PurchasePlanApplication {
    public static void main(String[] args) {
    SpringApplication.run(PurchasePlanApplication.class, args);}/*** 创建分布式ID对象* @return 分布式ID对象*/@Beanpublic IdGenerator getIdGenerator() {
    return new CommonSelfIdGenerator();}
}

2.3.6 配置类
　　　　和采购送货单服务相同，在此不贴了。
　　2.3.7 业务请求控制

@RestController
@RequestMapping("/purchaseplanweb")
public class PurchasePlanWeb {
    @Autowiredprivate PurchasePlanService purchasePlanService;@Autowiredprivate IdGenerator idGenerator;/*** 查询所有采购计划* @return 所有采购计划*/@RequestMapping("/query")public Object query() {
    return purchasePlanService.findByAll();}/*** 保存采购计划异常* @return*/@RequestMapping("/insertnormal")public Object insertNormal() {
    PurchasePlan purchasePlan = new PurchasePlan();//这里使用sharding的一个分布式ID组件，底层是使用snowflake算法实现。purchasePlan.setId(idGenerator.generateId().longValue());purchasePlan.setCode("测试采购计划编码" + purchasePlan.getId());purchasePlan.setName("测试采购计划名称" + purchasePlan.getId());purchasePlan.setNum(10D);purchasePlanService.insertNormal(purchasePlan);return "数据保存成功！";}/*** 保存采购计划异常* @return*/@RequestMapping("/insertexception")public Object insertException() {
    PurchasePlan purchasePlan = new PurchasePlan();//这里使用sharding的一个分布式ID组件，底层是使用snowflake算法实现。purchasePlan.setId(idGenerator.generateId().longValue());purchasePlan.setCode("测试采购计划编码" + purchasePlan.getId());purchasePlan.setName("测试采购计划名称" + purchasePlan.getId());purchasePlan.setNum(10D);purchasePlanService.insertException(purchasePlan);return "数据保存成功！";}
}

2.3.8 业务接口

public interface PurchasePlanService {
    /*** 查询所有采购计划* @return 采购计划集合*/public List<PurchasePlan> findByAll();/*** 保存采购计划正常* @param purchasePlan 采购计划*/public void insertNormal(PurchasePlan purchasePlan);/*** 保存采购计划异常* @param purchasePlan 采购计划*/public void insertException(PurchasePlan purchasePlan);/*** 更新采购计划* @param purchasePlan 采购计划*/public void update(PurchasePlan purchasePlan);/*** 删除采购计划* @param purchasePlan 采购计划*/public void delete(PurchasePlan purchasePlan);
}

2.3.9 业务接口实现

@Service
public class PurchasePlanServiceImpl implements PurchasePlanService {
    @Autowiredprivate PurchasePlanMapper purchasePlanMapper;@Autowiredprivate PurchaseDemandService purchaseDemandService;/*** 查询所有采购计划* @return 采购计划集合*/@Overridepublic List<PurchasePlan> findByAll() {
    return purchasePlanMapper.findByAll();}/*** 保存采购计划正常* @param purchasePlan 采购计划*/@GlobalTransactional(rollbackFor = Exception.class)@Overridepublic void insertNormal(PurchasePlan purchasePlan) {
    purchasePlanMapper.insert(purchasePlan);purchaseDemandService.insertNormal();}/*** 保存采购计划异常* @param purchasePlan 采购计划*/@Overridepublic void insertException(PurchasePlan purchasePlan) {
    purchasePlanMapper.insert(purchasePlan);purchaseDemandService.insertException();}/*** 更新采购计划* @param purchasePlan 采购计划*/@GlobalTransactional(rollbackFor = Exception.class)@Overridepublic void update(PurchasePlan purchasePlan) {
    purchasePlanMapper.update(purchasePlan);}/*** 删除采购计划* @param purchasePlan 采购计划*/@GlobalTransactional(rollbackFor = Exception.class)@Overridepublic void delete(PurchasePlan purchasePlan) {
    purchasePlanMapper.delete(purchasePlan);}
}

2.3.10 DAO接口

@Repository
public interface PurchasePlanMapper {
    /*** 查询所有采购计划* @return 采购计划集合*/public List<PurchasePlan> findByAll();/*** 保存采购计划* @param purchasePlan 采购计划*/public void insert(PurchasePlan purchasePlan);/*** 更新采购计划* @param purchasePlan 采购计划*/public void update(PurchasePlan purchasePlan);/*** 删除采购计划* @param purchasePlan 采购计划*/public void delete(PurchasePlan purchasePlan);
}

2.3.11 实体类

public class PurchasePlan implements Serializable {
    /****/private static final long serialVersionUID = 1L;private Long id; //主键private String code; //计划编码private String name; //计划名称private String type; //计划类型private Double num; //计划量public Long getId() {
    return id;}public void setId(Long id) {
    this.id = id;}public String getCode() {
    return code;}public void setCode(String code) {
    this.code = code;}public String getName() {
    return name;}public void setName(String name) {
    this.name = name;}public String getType() {
    return type;}public void setType(String type) {
    this.type = type;}public Double getNum() {
    return num;}public void setNum(Double num) {
    this.num = num;}
}

2.3.12 PurchasePlanMapper.xml

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
<mapper namespace="com.general.purchaseplan.mapper.PurchasePlanMapper" ><resultMap id="BaseResultMap" type="com.general.purchaseplan.entity.PurchasePlan"><id column="id" property="id" /><result column="code"  property="code" /><result column="name"  property="name" /><result column="num"  property="num" /></resultMap><select id="findByAll" resultMap="BaseResultMap">select * from purchase_plan</select><insert id="insert" parameterType="com.general.purchaseplan.entity.PurchasePlan">insert into purchase_plan(id, code, name, num) values(#{
    id}, #{
    code}, #{
    name}, #{
    num})</insert><update id="update" parameterType="com.general.purchaseplan.entity.PurchasePlan">update purchase_plan set code = #{
    code}, name = #{
    name}, num = #{
    num}where id = #{
    id};</update><delete id="delete" parameterType="com.general.purchaseplan.entity.PurchasePlan">delete from purchase_plan where id = #{
    id};</delete>
</mapper>

2.3.13 采购需求feign调用接口

@Component
@FeignClient(value = "PURCHASE-DEMAND")
public interface PurchaseDemandService {
    /*** 保存采购需求正常* @return 处理信息*/@GetMapping(value = "/purchasedemandweb/insertnormal")public String insertNormal();/*** 保存采购需求异常* @return 处理信息*/@GetMapping(value = "/purchasedemandweb/insertexception")public String insertException();
}

2.4 采购需求
　　2.4.1 工程结构

2.4.2 pom文件依赖包
　　　　和采购送货单服务相同，在此不贴了。
　　2.4.3 application.properties文件配置

#服务端口
server.port=8082
#tomcat编码
server.tomcat.uri-encoding=UTF-8
spring.aop.proxy-target-class=true
spring.application.name=purchase-demand
spring.cloud.alibaba.seata.tx-service-group=purchase-demand-group
spring.main.allow-bean-definition-overriding=true
#表示是否将自己注册进EurekaServer默认为true
eureka.client.fetch-registry=true
#是否从EurekaServer抓取已有的注册信息，默认为true。单节点无所谓，集群必须设置为true才能配合ribbon使用负载均衡
eureka.client.register-with-eureka=true
#eureka服务器地址单机
eureka.client.service-url.defaultZone=http://127.0.0.1:8181/eureka/
#注册到eureka中的服务名称
eureka.instance.instance-id=purchase-demand
#访问路径可以显示IP地址
eureka.instance.prefer-ip-address=true
#feign客户端读取超时时间(单位：毫秒)
feign.client.config.default.read-timeout=15000
#feign客户端连接超时时间(单位：毫秒)
feign.client.config.default.connect-timeout=15000
#当OkToRetryOnAllOperations设置为false时，只会对get请求进行重试。如果设置为true，便会对所有的请求进行重试，如果是put或post等写操作，如果服务器接口没做幂等性，会产生不好的结果，所以OkToRetryOnAllOperations慎用。
ribbon.OkToRetryOnAllOperations=false
#数据源名称，多数据源以逗号分隔
spring.shardingsphere.datasource.names=master,slave0,slave1
#数据库连接池类名称
spring.shardingsphere.datasource.master.type=com.alibaba.druid.pool.DruidDataSource
#数据库驱动类名
spring.shardingsphere.datasource.master.driver-class-name=com.mysql.cj.jdbc.Driver
#数据库URL连接
spring.shardingsphere.datasource.master.url=jdbc:mysql://192.168.145.183:3306/order?useUnicode=true&characterEncoding=utf8&tinyInt1isBit=false&useSSL=false&serverTimezone=GMT
#数据库用户名
spring.shardingsphere.datasource.master.username=root
#数据库密码
spring.shardingsphere.datasource.master.password=123456
#初始化大小，最小，最大
spring.shardingsphere.datasource.master.initial-size=5
spring.shardingsphere.datasource.master.min-idle=5
spring.shardingsphere.datasource.master.max-active=20
#配置获取连接等待超时的时间
spring.shardingsphere.datasource.master.max-wait=60000
spring.shardingsphere.datasource.slave0.type=com.alibaba.druid.pool.DruidDataSource
spring.shardingsphere.datasource.slave0.driver-class-name=com.mysql.cj.jdbc.Driver
spring.shardingsphere.datasource.slave0.url=jdbc:mysql://192.168.145.184:3306/order?useUnicode=true&characterEncoding=utf8&tinyInt1isBit=false&useSSL=false&serverTimezone=GMT
spring.shardingsphere.datasource.slave0.username=root
spring.shardingsphere.datasource.slave0.password=123456
#初始化大小，最小，最大
spring.shardingsphere.datasource.slave0.initial-size=5
spring.shardingsphere.datasource.slave0.min-idle=5
spring.shardingsphere.datasource.slave0.max-active=20
#配置获取连接等待超时的时间
spring.shardingsphere.datasource.slave0.max-wait=60000
spring.shardingsphere.datasource.slave1.type=com.alibaba.druid.pool.DruidDataSource
spring.shardingsphere.datasource.slave1.driver-class-name=com.mysql.cj.jdbc.Driver
spring.shardingsphere.datasource.slave1.url=jdbc:mysql://192.168.145.185:3306/order?useUnicode=true&characterEncoding=utf8&tinyInt1isBit=false&useSSL=false&serverTimezone=GMT
spring.shardingsphere.datasource.slave1.username=root
spring.shardingsphere.datasource.slave1.password=123456
#初始化大小，最小，最大
spring.shardingsphere.datasource.slave1.initial-size=5
spring.shardingsphere.datasource.slave1.min-idle=5
spring.shardingsphere.datasource.slave1.max-active=20
#配置获取连接等待超时的时间
spring.shardingsphere.datasource.slave1.max-wait=60000
#读写分离数据源名称
spring.shardingsphere.masterslave.name=ms
#主数据源名称
spring.shardingsphere.masterslave.master-data-source-name=master
#从数据源名称，多个从数据源用逗号分隔
spring.shardingsphere.masterslave.slave-data-source-names=slave0,slave1
#负载均衡算法名称
spring.shardingsphere.masterslave.load-balance-algorithm-type=round_robin
#开启SQL显示
spring.shardingsphere.props.sql.show=true
#mybatisplus配置mapper的xml位置
mybatis-plus.mapper-locations=classpath:mybatis/mapper/**/*.xml
#mybatisplus配置别名扫描
mybatis-plus.typeAliasesPackage=com.general.**.entity
logging.level.root=info

2.4.4 seata配置文件
　　1.registry.conf

registry {
    # file 、nacos 、eureka、redis、zk、consul、etcd3、sofatype = "file"file {
    name = "file.conf"}
}config {
    # file、nacos 、apollo、zk、consul、etcd3、springCloudConfigtype = "file"file {
    name = "file.conf"}
}

2.file.conf

transport {
    type = "TCP"server = "NIO"heartbeat = trueenableClientBatchSendRequest = truethreadFactory {
    bossThreadPrefix = "NettyBoss"workerThreadPrefix = "NettyServerNIOWorker"serverExecutorThread-prefix = "NettyServerBizHandler"shareBossWorker = falseclientSelectorThreadPrefix = "NettyClientSelector"clientSelectorThreadSize = 1clientWorkerThreadPrefix = "NettyClientWorkerThread"bossThreadSize = 1workerThreadSize = "default"}shutdown {
    wait = 3}serialization = "seata"compressor = "none"
}
service {
    vgroupMapping.purchase-demand-group = "seata-server"seata-server.grouplist = "192.168.145.185:8091"enableDegrade = falsedisableGlobalTransaction = false
}client {
    rm {
    asyncCommitBufferLimit = 10000lock {
    retryInterval = 10retryTimes = 30retryPolicyBranchRollbackOnConflict = true}reportRetryCount = 5tableMetaCheckEnable = falsereportSuccessEnable = false}tm {
    commitRetryCount = 5rollbackRetryCount = 5}undo {
    dataValidation = truelogSerialization = "jackson"logTable = "undo_log"}log {
    exceptionRate = 100}
}

3.seata.conf

client {
    application.id = purchase-demandtransaction.service.group = purchase-demand-group
}

2.4.5 启动类

@EnableEurekaClient
@EnableFeignClients
@MapperScan("com.general.**.mapper")
@ComponentScan(value = "com.general")
@SpringBootApplication(exclude = {
     DruidDataSourceAutoConfigure.class, DataSourceAutoConfiguration.class })
public class PurchaseDemandApplication {
    public static void main(String[] args) {
    SpringApplication.run(PurchaseDemandApplication.class, args);}/*** 创建分布式ID对象* @return 分布式ID对象*/@Beanpublic IdGenerator getIdGenerator() {
    return new CommonSelfIdGenerator();}
}

2.4.6 配置类
　　　　和采购送货单服务相同，在此不贴了。
　　2.4.7 业务请求控制

@RestController
@RequestMapping("/purchasedemandweb")
public class PurchaseDemandWeb {
    @Autowiredprivate PurchaseDemandService purchaseDemandService;@Autowiredprivate IdGenerator idGenerator;/*** 查询所有请求控制* @return 所有请求控制*/@RequestMapping("/query")public Object query() {
    return purchaseDemandService.findByAll();}/*** 保存请求控制* @return*/@RequestMapping("/insertnormal")public Object insertNormal() {
    PurchaseDemand purchaseDemand = new PurchaseDemand();//这里使用sharding的一个分布式ID组件，底层是使用snowflake算法实现。purchaseDemand.setId(idGenerator.generateId().longValue());purchaseDemand.setCode("测试采购需求编码" + purchaseDemand.getId());purchaseDemand.setName("测试采购需求名称" + purchaseDemand.getId());purchaseDemand.setNum(10D);purchaseDemandService.insertNormal(purchaseDemand);return "数据保存成功！";}/*** 保存请求控制* @return*/@RequestMapping("/insertexception")public Object insertException() {
    PurchaseDemand purchaseDemand = new PurchaseDemand();//这里使用sharding的一个分布式ID组件，底层是使用snowflake算法实现。purchaseDemand.setId(idGenerator.generateId().longValue());purchaseDemand.setCode("测试采购需求编码" + purchaseDemand.getId());purchaseDemand.setName("测试采购需求名称" + purchaseDemand.getId());purchaseDemand.setNum(10D);purchaseDemandService.insertException(purchaseDemand);return "数据保存成功！";}
}

2.4.8 业务接口

public interface PurchaseDemandService {
    /*** 查询所有采购需求* @return 采购需求集合*/public List<PurchaseDemand> findByAll();/*** 保存采购需求* @param purchaseDemand*/public void insertNormal(PurchaseDemand purchaseDemand);/*** 保存采购需求* @param purchaseDemand*/public void insertException(PurchaseDemand purchaseDemand);/*** 更新采购需求* @param purchaseDemand*/public void update(PurchaseDemand purchaseDemand);/*** 删除采购需求* @param purchaseDemand*/public void delete(PurchaseDemand purchaseDemand);
}

2.4.9 业务接口实现

@Service
public class PurchaseDemandServiceImpl implements PurchaseDemandService {
    @Autowiredprivate PurchaseDemandMapper purchaseDemandMapper;/*** 查询所有采购需求* @return 采购需求集合*/@Overridepublic List<PurchaseDemand> findByAll() {
    return purchaseDemandMapper.findByAll();}/*** 保存采购需求正常* @param purchaseDemand*/@GlobalTransactional(rollbackFor = Exception.class)@Overridepublic void insertNormal(PurchaseDemand purchaseDemand) {
    purchaseDemandMapper.insert(purchaseDemand);}/*** 保存采购需求异常* @param purchaseDemand*/@GlobalTransactional(rollbackFor = Exception.class)@Overridepublic void insertException(PurchaseDemand purchaseDemand) {
    purchaseDemandMapper.insert(purchaseDemand);throw new RuntimeException(purchaseDemand.getId() + "采购需求保存异常");}/*** 更新采购需求* @param purchaseDemand*/@GlobalTransactional(rollbackFor = Exception.class)@Overridepublic void update(PurchaseDemand purchaseDemand) {
    purchaseDemandMapper.update(purchaseDemand);}/*** 删除采购需求* @param purchaseDemand*/@GlobalTransactional(rollbackFor = Exception.class)@Overridepublic void delete(PurchaseDemand purchaseDemand) {
    purchaseDemandMapper.delete(purchaseDemand);}
}

2.4.10 DAO接口

@Repository
public interface PurchaseDemandMapper {
    /*** 查询所有采购需求* @return*/public List<PurchaseDemand> findByAll();/*** 保存采购需求* @param purchaseDemand 采购需求*/public void insert(PurchaseDemand purchaseDemand);/*** 更新采购需求* @param purchaseDemand 采购需求*/public void update(PurchaseDemand purchaseDemand);/*** 删除采购需求* @param purchaseDemand 采购需求*/public void delete(PurchaseDemand purchaseDemand);
}

2.4.11 实体类

public class PurchaseDemand implements Serializable {
    /****/private static final long serialVersionUID = 1L;private Long id; //主键private String code; //需求编码private String name; //需求名称private Double num; //需求量public Long getId() {
    return id;}public void setId(Long id) {
    this.id = id;}public String getCode() {
    return code;}public void setCode(String code) {
    this.code = code;}public String getName() {
    return name;}public void setName(String name) {
    this.name = name;}public Double getNum() {
    return num;}public void setNum(Double num) {
    this.num = num;}
}

2.4.12 PurchaseDemandMapper.xml

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
<mapper namespace="com.general.purchasedemand.mapper.PurchaseDemandMapper" ><resultMap id="BaseResultMap" type="com.general.purchasedemand.entity.PurchaseDemand"><id column="id" property="id" /><result column="code"  property="code" /><result column="name"  property="name" /><result column="num"  property="num" /></resultMap><select id="findByAll" resultMap="BaseResultMap">select * from purchase_demand</select><insert id="insert" parameterType="com.general.purchasedemand.entity.PurchaseDemand">insert into purchase_demand(id, code, name, num) values(#{
    id}, #{
    code}, #{
    name}, #{
    num})</insert><update id="update" parameterType="com.general.purchasedemand.entity.PurchaseDemand">update purchase_demand set code = #{
    code}, name = #{
    name}, num = #{
    num}where id = #{
    id};</update><delete id="delete" parameterType="com.general.purchasedemand.entity.PurchaseDemand">delete from purchase_demand where id = #{
    id};</delete>  
</mapper>

2.5 注册服务
　　2.5.1 工程结构

2.5.2 pom文件依赖包

<dependencies><!-- spring boot包引入 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope><exclusions><exclusion><groupId>org.junit.vintage</groupId><artifactId>junit-vintage-engine</artifactId></exclusion></exclusions></dependency><!--eureka服务端依赖包引入--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-server</artifactId><version>2.2.1.RELEASE</version></dependency></dependencies>

2.5.3 application.properties文件配置

#服务端口
server.port=8181
#tomcat编码
server.tomcat.uri-encoding=UTF-8
#日志级别
logging.level.root=info
#eureka服务端的实例名称
eureka.instance.hostname=eureka8181.com
#false表示不向注册中心注册自己
eureka.client.register-with-eureka=false
#false表示自己端就是注册中心，我的职责就是维护服务实例，并不需要去检索服务
eureka.client.fetch-registry=false
#单机就是自己，如果是集群指向其它eureka
eureka.client.service-url.defaultZone=http://127.0.0.1:8181/eureka/

2.5.4 启动类

@SpringBootApplication
@EnableEurekaServer
public class EurekaServerApplication {
    public static void main(String[] args) {
    SpringApplication.run(EurekaServerApplication.class, args);}
}

2.6 测试
　　2.6.1 正常业务测试
　　分别启动注册中心、采购送货单、采购计划、采购需求服务。在浏览器里输入：
http://127.0.0.1:8081/deliverynoteweb/insertnormal，刷新三次。
能看到过程如下图：
插入数据前端

数据库数据：

后台日志：

2.6.2 异常业务测试
　　在浏览器里输入：http://127.0.0.1:8081/deliverynoteweb/insertexception
看到结果如下图：
前端响应如下：

后台日志：

数据库数据：

到此整个分布式事务已经全部实现。