分布式事务两阶段提交——Eureka+Seata方案

分布式事务两阶段提交——Nacos+Seata方案

前言

在微服务的大环境下，服务按照业务维度拆分之后会遇到事务不一致问题，Seata的开源填补了两阶段提交这种模式，并且无业务代码的侵入，这里采用eureka集群整合Seata。

一、Eureka集群搭建

1、修改hosts文件映射

127.0.0.1           eureka-server1.com
127.0.0.1           eureka-server2.com

2、创建eureka-server工程，引入Maven依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.10.RELEASE</version>
    <relativePath/>
</parent>

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-netflix-eureka-server</artifactId>
    </dependency>
</dependencies>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>Hoxton.SR8</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
        </plugin>
    </plugins>
</build>

3、application.properties文件

• 1)application.properties

  
  

spring.profiles.active=eureka-server1

• 2)application-eureka-server1.properties

# Eureka Server服务端口
server.port=9090

# 取消服务器自我注册，就是Eureka Server也可以被更高层的服务器来管理
eureka.client.register-with-eureka=false
# 注册中心的服务器，没有必要再去检索服务
eureka.client.fetch-registry=false

# 单机 hostname: localhost #eureka注册中心实例名称
eureka.instance.hostname=eureka-server1.com

# Eureka Server 服务URL，用于客户端注册
#设置与Eureka注册中心交互的地址，查询服务和注册服务用到
#集群
eureka.client.service-url.defaultZone=http://eureka-server2.com:9091/eureka/

#单机
#eureka.client.serverUrl.defaultZone=http://localhost:${server.port}/eureka/

• 3)application-eureka-server2.properties

# Eureka Server服务端口
server.port=9091


# 取消服务器自我注册，就是Eureka Server也可以被更高层的服务器来管理
eureka.client.register-with-eureka=false
# 注册中心的服务器，没有必要再去检索服务
eureka.client.fetch-registry=false

# 单机 hostname: localhost #eureka注册中心实例名称
eureka.instance.hostname=eureka-server2.com

# Eureka Server 服务URL，用于客户端注册
#设置与Eureka注册中心交互的地址，查询服务和注册服务用到
#集群
eureka.client.service-url.defaultZone=http://eureka-server1.com:9090/eureka/

#单机
#eureka.client.serverUrl.defaultZone=http://localhost:${server.port}/eureka/

注意，多台eureka-server服务，只需要修改eureka.instance.hostname和eureka.client.service-url.defaultZone

4、新建EurekaServerApplication启动类

@SpringBootApplication
@EnableEurekaServer
public class EurekaServerApplication {

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

5、启动eureka-server服务

二、Seata配置

2.1、Seata服务端（TC）部署

下载Seata服务端压缩包：https://github.com/seata/seata/releases

2.2、Seata配置

1、修改conf目录中 flie.conf 文件，修改事务日志存储模式为 db 及数据库连接信息，且新增service模块，如下：

transport {
  # tcp udt unix-domain-socket
  type = "TCP"
  #NIO NATIVE
  server = "NIO"
  #enable heartbeat
  heartbeat = true
  # the client batch send request enable
  enableClientBatchSendRequest = false
  #thread factory for netty
  threadFactory {
    bossThreadPrefix = "NettyBoss"
    workerThreadPrefix = "NettyServerNIOWorker"
    serverExecutorThreadPrefix = "NettyServerBizHandler"
    shareBossWorker = false
    clientSelectorThreadPrefix = "NettyClientSelector"
    clientSelectorThreadSize = 1
    clientWorkerThreadPrefix = "NettyClientWorkerThread"
    # netty boss thread size,will not be used for UDT
    bossThreadSize = 1
    #auto default pin or 8
    workerThreadSize = "default"
  }
  shutdown {
    # when destroy server, wait seconds
    wait = 3
  }
  serialization = "seata"
  compressor = "none"
}

#这里手动加入service模块
service {
  #transaction service group mapping
  #修改，可不改,my_test_tx_group 前缀建议为各微服务名。
  vgroup_mapping.seata_eureka_bank1_group = "seata-server"
  vgroup_mapping.seata_eureka_bank2_group = "seata-server"
  #only support when registry.type=file, please don't set multiple addresses
  # 此服务的地址
  default.grouplist = "127.0.0.1:8091"
  #disable seata
  disableGlobalTransaction = false
}

## transaction log store, only used in server side
store {
  ## store mode: file、db
  mode = "db"
  ## file store property
  file {
    ## store location dir
    dir = "sessionStore"
    # branch session size , if exceeded first try compress lockkey, still exceeded throws exceptions
    maxBranchSessionSize = 16384
    # globe session size , if exceeded throws exceptions
    maxGlobalSessionSize = 512
    # file buffer size , if exceeded allocate new buffer
    fileWriteBufferCacheSize = 16384
    # when recover batch read size
    sessionReloadReadSize = 100
    # async, sync
    flushDiskMode = async
  }

  ## database store property
  db {
    ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.
    datasource = "druid"
    ## mysql/oracle/postgresql/h2/oceanbase etc.
    dbType = "mysql"
    driverClassName = "com.mysql.jdbc.Driver"
    url = "jdbc:mysql://localhost:3306/seata"
    user = "root"
    password = "yibo"
    minConn = 5
    maxConn = 30
    globalTable = "global_table"
    branchTable = "branch_table"
    lockTable = "lock_table"
    queryLimit = 100
  }
}
## server configuration, only used in server side
server {
  recovery {
    #schedule committing retry period in milliseconds
    committingRetryPeriod = 1000
    #schedule asyn committing retry period in milliseconds
    asynCommittingRetryPeriod = 1000
    #schedule rollbacking retry period in milliseconds
    rollbackingRetryPeriod = 1000
    #schedule timeout retry period in milliseconds
    timeoutRetryPeriod = 1000
  }
  undo {
    logSaveDays = 7
    #schedule delete expired undo_log in milliseconds
    logDeletePeriod = 86400000
  }
  #check auth
  enableCheckAuth = true
  #unit ms,s,m,h,d represents milliseconds, seconds, minutes, hours, days, default permanent
  maxCommitRetryTimeout = "-1"
  maxRollbackRetryTimeout = "-1"
  rollbackRetryTimeoutUnlockEnable = false
}

## metrics configuration, only used in server side
metrics {
  enabled = false
  registryType = "compact"
  # multi exporters use comma divided
  exporterList = "prometheus"
  exporterPrometheusPort = 9898
}

由于我们使用了db模式存储事务日志，所以我们需要创建一个seata数据库，Seata数据库表初始化脚本：https://github.com/seata/seata/tree/1.1.0/script/server/db

2.3、修改注册中心和配置中心，使用eureka作为注册中心、直接使用file.conf配置文件存储seata规则，即修改 conf目录中 registry.conf 文件，如下：

registry {
  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  type = "eureka"

  nacos {
    application = "seata-server"
    serverAddr = "127.0.0.1:8848"
    group = "SEATA_GROUP"
    namespace = ""
    cluster = "default"
    username = ""
    password = ""
  }
  eureka {
    serviceUrl = "http://eureka-server1.com:9090/eureka/,http://eureka-server2.com:9091/eureka/"
    application = "seata-server"
    weight = "1"
  }
  redis {
    serverAddr = "localhost:6379"
    db = 0
    password = ""
    cluster = "default"
    timeout = 0
  }
  zk {
    cluster = "default"
    serverAddr = "127.0.0.1:2181"
    sessionTimeout = 6000
    connectTimeout = 2000
    username = ""
    password = ""
  }
  consul {
    cluster = "default"
    serverAddr = "127.0.0.1:8500"
  }
  etcd3 {
    cluster = "default"
    serverAddr = "http://localhost:2379"
  }
  sofa {
    serverAddr = "127.0.0.1:9603"
    application = "default"
    region = "DEFAULT_ZONE"
    datacenter = "DefaultDataCenter"
    cluster = "default"
    group = "SEATA_GROUP"
    addressWaitTime = "3000"
  }
  file {
    name = "file.conf"
  }
}

config {
  # file、nacos 、apollo、zk、consul、etcd3
  type = "file"

  nacos {
    serverAddr = "127.0.0.1:8848"
    namespace = ""
    group = "SEATA_GROUP"
    username = ""
    password = ""
  }
  consul {
    serverAddr = "127.0.0.1:8500"
  }
  apollo {
    appId = "seata-server"
    apolloMeta = "http://192.168.1.204:8801"
    namespace = "application"
  }
  zk {
    serverAddr = "127.0.0.1:2181"
    sessionTimeout = 6000
    connectTimeout = 2000
    username = ""
    password = ""
  }
  etcd3 {
    serverAddr = "http://localhost:2379"
  }
  file {
    name = "file.conf"
  }
}

2.4、启动seata-server，如下：

三、各微服务配置

3.1、引入maven依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.10.RELEASE</version>
    <relativePath/>
</parent>

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
    </dependency>

    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-alibaba-seata</artifactId>
        <version>2.2.0.RELEASE</version>
        <exclusions>
            <exclusion>
                <groupId>io.seata</groupId>
                <artifactId>seata-all</artifactId>
            </exclusion>
            <exclusion>
                <groupId>io.seata</groupId>
                <artifactId>seata-spring-boot-starter</artifactId>
            </exclusion>
        </exclusions>
    </dependency>

    <dependency>
        <groupId>io.seata</groupId>
        <artifactId>seata-spring-boot-starter</artifactId>
        <version>1.3.0</version>
    </dependency>

    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-openfeign</artifactId>
    </dependency>

    <dependency>
        <groupId>tk.mybatis</groupId>
        <artifactId>mapper-spring-boot-starter</artifactId>
        <version>2.1.5</version>
    </dependency>

    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.18</version>
    </dependency>

    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.12</version>
        <scope>provided</scope>
    </dependency>
</dependencies>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>Hoxton.SR8</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
        </plugin>

        <plugin>
            <groupId>org.mybatis.generator</groupId>
            <artifactId>mybatis-generator-maven-plugin</artifactId>
            <version>1.3.6</version>
            <configuration>
                <configurationFile>
                    ${basedir}/src/main/resources/generator/generatorConfig.xml
                </configurationFile>
                <overwrite>true</overwrite>
                <verbose>true</verbose>
            </configuration>
            <dependencies>
                <dependency>
                    <groupId>mysql</groupId>
                    <artifactId>mysql-connector-java</artifactId>
                    <version>8.0.18</version>
                </dependency>
                <dependency>
                    <groupId>tk.mybatis</groupId>
                    <artifactId>mapper</artifactId>
                    <version>4.1.5</version>
                </dependency>
            </dependencies>
        </plugin>
    </plugins>
</build>    

3.2、分别在各业务数据库中创建undo_log表，此表为seata框架使用，sql地址：https://github.com/seata/seata/tree/develop/script/client/at/db

3.3、配置application.properties文件

# 应用名
spring.application.name=eureka-seata-bank1

server.port=8080

#表示是否将自己注册进EurekaServer默认为true
eureka.client.register-with-eureka=true
#是否从EurekaServer抓取已有的注册信息，默认为true，单节点无所谓，集群必须设置为true才能配合ribbon使用负载均衡
eureka.client.fetch-registry=true
#集群版
eureka.client.service-url.defaultZone=http://eureka-server1.com:9090/eureka/,http://eureka-server2.com:9091/eureka/

# Eureka 客户端应用实例的ID
eureka.instance.instance-id=${spring.application.name}:${server.port}
#点进去左下角会显示ip
eureka.instance.prefer-ip-address=true

# 调整注册信息的获取周期
eureka.client.registry-fetch-interval-seconds=5

# 调整客户端应用状态信息上报的周期
eureka.client.instance-info-replication-interval-seconds=5

# seata config.type=file相关配置
seata.enabled=true
seata.application-id=${spring.application.name}

# 不同的微服务vgroup_mapping.seata_eureka_bank1_group配置不同
#这里的名字与file.conf中vgroup_mapping.seata_eureka_bank1_group = "seata-server"相同
seata.tx-service-group=seata_eureka_bank1_group
#这里的名字与file.conf中vgroup_mapping.seata_eureka_bank1_group = "seata-server"相同
seata.service.vgroup-mapping.seata_eureka_bank1_group=seata-server
#这里的名字与file.conf中default.grouplist = "127.0.0.1:8091"相同
seata.service.grouplist.default=127.0.0.1:8091
# 开启数据源自动代理
seata.enable-auto-data-source-proxy=true
# 配置中心为本地file文件
seata.config.type=file

# 配置中心为本地file文件的文件名称
seata.config.file.name=file.conf

seata.registry.type=eureka
seata.registry.eureka.application=seata-server
seata.registry.eureka.serviceUrl=http://eureka-server1.com:9090/eureka/,http://eureka-server2.com:9091/eureka/
seata.registry.eureka.weight=1

mybatis.type-aliases-package=com.yibo.eureka.seata.entity
mybatis.mapper-locations=classpath:mapper/*.xml
mapper.identity=MYSQL
mapper.not-empty=false

spring.datasource.url=jdbc:mysql://localhost:3306/trade_bank1?useUnicode=true&useAffectedRows=true&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=UTC
spring.datasource.username=root
spring.datasource.password=yibo
spring.datasource.driver-class-name=com.mysql.jdbc.Driver

# 设置连接超时时间 default 2000
ribbon.ConnectTimeout=6000
# 设置读取超时时间  default 5000
ribbon.ReadTimeout=6000
# 对所有操作请求都进行重试  default false
ribbon.OkToRetryOnAllOperations=true
# 切换实例的重试次数  default 1
ribbon.MaxAutoRetriesNextServer=2
# 对当前实例的重试次数 default 0
ribbon.MaxAutoRetries=1

3.4、启动类配置

@MapperScan("com.yibo.eureka.seata.mapper")//扫描mybatis的指定包下的接口
@SpringBootApplication
@EnableDiscoveryClient  //服务发现，对外暴露服务
@EnableEurekaClient     //本服务启动后会自动注册进Eureka服务中
@EnableFeignClients
public class EurekaSeataBank1Application {

    public static void main(String[] args) {
        SpringApplication.run(EurekaSeataBank1Application.class,args);
    }
}

四、业务逻辑实现

4.1、Controller实现

@RestController
@RequestMapping("/bank1")
public class Bank1Controller {

    @Autowired
    private AccountService accountService;
    
    @GetMapping("/transfer/{amount}")
    public String transfer(@PathVariable("amount") Long amount){
        accountService.updateAccountBalance("1",amount);
        return "bank1"+amount;
    }
}

4.2、Service实现，@GlobalTransactional注解用以开启全局事务，@Transactional注解用于分支事务

@Service
@Slf4j
public class AccountServiceImpl implements AccountService {

    @Autowired
    private AccountInfoMapper accountInfoMapper;

    @Autowired
    private Bank2Client bank2Client;

    @Transactional
    @GlobalTransactional//开启全局事务
    public void updateAccountBalance(String accountNo, Long amount) {
        log.info("bank1 service begin,XID：{}", RootContext.getXID());
        //扣减张三的金额
        accountInfoMapper.updateAccountBalance(accountNo,amount *-1);
        //调用李四微服务，转账
        String transfer = bank2Client.transfer(amount);
        if("fallback".equals(transfer)){
            //调用李四微服务异常
            throw new RuntimeException("调用李四微服务异常");
        }
        if(amount == 2){
            //人为制造异常
            throw new RuntimeException("bank1 make exception..");
        }
    }
}

4.3、Bank2Client接口的FeignClient

@FeignClient(value="eureka-seata-bank2")
public interface Bank2Client {

    //远程调用微服务
    @GetMapping("/bank2/transfer/{amount}")
    public String transfer(@PathVariable("amount") Long amount);
}

其他微服务按此配置即可。

github源码地址：https://github.com/jjhyb/distributed-transaction