Seata分布式事务框架Sample

前言

阿里官方给出了seata-sample地址，官方自己也对Sample提供了很多类型，可以查看学习。我这里选择演示SpringBoot+MyBatis。

该聚合工程共包括5个module：

sbm-account-service
sbm-business-service
sbm-common-service
sbm-order-service
sbm-storage-service

不同Module之间的服务通信使用Rest方式通信。

准备工作

创建数据库

在sql/all_in_one_sql里是演示中需要的sql脚本，共创建了3个schema: db_account, db_order, db_storage，分别演示三个不同的数据库，每个schema里都有一张undo_log表。

启动Seata-Server

Seata-Server扮演TM的角色，在官网下载http://seata.io/zh-cn/blog/download.html，最新版本为1.4.2。

/conf/registry.conf 中有两个大的节点registry - 注册中心配置选项，config - 配置中心配置选项。

# 注册中心配置

registry {

  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa

  type = "file"

  nacos {

    application = "seata-server"

    serverAddr = "127.0.0.1:8848"

    group = "SEATA_GROUP"

    namespace = ""

    cluster = "default"

    username = ""

    password = ""

  }

  eureka {

    serviceUrl = "http://localhost:8761/eureka"

    application = "default"

    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"

    aclToken = ""

  }

  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 = ""

    dataId = "seataServer.properties"

  }

  consul {

    serverAddr = "127.0.0.1:8500"

    aclToken = ""

  }

  apollo {

    appId = "seata-server"

    ## apolloConfigService will cover apolloMeta

    apolloMeta = "http://192.168.1.204:8801"

    apolloConfigService = "http://192.168.1.204:8080"

    namespace = "application"

    apolloAccesskeySecret = ""

    cluster = "seata"

  }

  zk {

    serverAddr = "127.0.0.1:2181"

    sessionTimeout = 6000

    connectTimeout = 2000

    username = ""

    password = ""

    nodePath = "/seata/seata.properties"

  }

  etcd3 {

    serverAddr = "http://localhost:2379"

  }

  file {

    name = "file.conf"

  }

}

/conf/file.conf - 只有当registry.conf下 config.type=file时才加载file.config中的参数。config.type等于其他值的话则不需要file.config。 seata-server也提供了file.conf.example，详细的参数介绍也可以查看http://seata.io/zh-cn/docs/user/configurations.html。

## transaction log store, only used in seata-server

store {

  ## store mode: file、db、redis

  mode = "file"

  ## rsa decryption public key

  publicKey = ""

  ## 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)/HikariDataSource(hikari) etc.

    datasource = "druid"

    ## mysql/oracle/postgresql/h2/oceanbase etc.

    dbType = "mysql"

    driverClassName = "com.mysql.jdbc.Driver"

    ## if using mysql to store the data, recommend add rewriteBatchedStatements=true in jdbc connection param

    url = "jdbc:mysql://127.0.0.1:3306/seata?rewriteBatchedStatements=true"

    user = "mysql"

    password = "mysql"

    minConn = 5

    maxConn = 100

    globalTable = "global_table"

    branchTable = "branch_table"

    lockTable = "lock_table"

    queryLimit = 100

    maxWait = 5000

  }

  ## redis store property

  redis {

    ## redis mode: single、sentinel

    mode = "single"

    ## single mode property

    single {

      host = "127.0.0.1"

      port = "6379"

    }

    ## sentinel mode property

    sentinel {

      masterName = ""

      ## such as "10.28.235.65:26379,10.28.235.65:26380,10.28.235.65:26381"

      sentinelHosts = ""

    }

    password = ""

    database = "0"

    minConn = 1

    maxConn = 10

    maxTotal = 100

    queryLimit = 100

  }

}

启动服务./bin/seata-server.bat ，默认打开了8091端口。

启动Sample服务

依此启动account-service，business-service，order-service，storage-service。business-service作为业务逻辑的入口分别调用order-service和storage-service。这里就看到了关键注解GlobalTransactional。

4个服务的默认端口分别是8081，8082， 8083，8084。

BusinessService

@Service

public class BusinessService {

    private static final Logger LOGGER = LoggerFactory.getLogger(BusinessService.class);

    @Autowired

    private StorageClient storageClient;

    @Autowired

    private OrderClient orderClient;

    /**

     * 减库存，下订单

     *

     * @param userId

     * @param commodityCode

     * @param orderCount

     */

    @GlobalTransactional

    public void purchase(String userId, String commodityCode, int orderCount) {

        LOGGER.info("purchase begin ... xid: " + RootContext.getXID());

        storageClient.deduct(commodityCode, orderCount);

        orderClient.create(userId, commodityCode, orderCount);

    }

}

OrderClient

@Slf4j

@Component

public class OrderClient {

    @Autowired

    private RestTemplate restTemplate;

    public void create(String userId, String commodityCode, int orderCount) {

        String url = "http://127.0.0.1:8082/api/order/debit?userId=" + userId + "&commodityCode=" + commodityCode + "&count=" + orderCount;

        try {

            restTemplate.getForEntity(url, Void.class);

        } catch (Exception e) {

            log.error("create url {} ,error:", url);

            throw new RuntimeException();

        }

    }

}

StorageClient

@Slf4j

@Component

public class StorageClient {

    @Autowired

    private RestTemplate restTemplate;

    public void deduct(String commodityCode, int orderCount) {

        System.out.println("business to storage " + RootContext.getXID());

        String url = "http://127.0.0.1:8081/api/storage/deduct?commodityCode=" + commodityCode + "&count=" + orderCount;

        try {

            restTemplate.getForEntity(url, Void.class);

        } catch (Exception e) {

            log.error("deduct url {} ,error:", url, e);

            throw new RuntimeException();

        }

    }

}

运行Sample

模拟正常事务提交

运行Sample前，我们先查看下当前三个服务数据库的数据。

db_account.account_tbl

db_storage.storage_tbl

业务接口在business-service/BusinessController里，我们先来执行下购买下单正常的提交流程。

执行接口http://localhost:8094/api/business/purchase/commit/, 执行后，seata-server控制台上会显示全局事务执行的具体日志和执行成功的日志。

执行后我们再查看下数据库。

db_account.account_tbl。 user_id对应1001的账户减去了5元（9995）。

db_storage.storage_tbl

db_order.order_tbl