前言

seata是阿里巴巴研发的一套开源分布式事务框架,提供了AT、TCC、SAGA 和 XA 几种事务模式。本文以精品课项目组的物流后盾服务为例,介绍seata框架落地的过程,遇到的问题以及解决方案。

作者/ 邓新伟

编辑/ 网易有道

有道精品课教务零碎是基于springcloud的分布式集群服务。在理论业务中,存在许多分布式事务场景。然而传统的事务框架是无奈实现全局事务的。长期以来,咱们的分布式场景的一致性,往往指的是放弃强一致性,保障最终一致性。

咱们从调研中发现,seata框架既能够满足业务需要,灵便兼容多种事务模式,又能够实现数据强一致性。

本文以物流业务为例,记录了在理论业务中落地seata框架落地的过程中遇到的一些问题以及解决方案,供大家学习探讨~欢送大家在留言区探讨交换

1. 根底信息

  • seata版本:1.4
  • 微服务框架:springcloud
  • 注册核心:consul

2.根本框架

2.1 根本组件

seata框架分为3个组件:

  • TC (Transaction Coordinator) -事务协调者 (即seata-server)

保护全局和分支事务的状态,驱动全局事务提交或回滚。

  • TM (Transaction Manager) -事务管理器 (在client上,发动事务的服务)

定义全局事务的范畴:开始全局事务、提交或回滚全局事务。

  • RM (Resource Manager) - 资源管理器 (在client)

治理分支事务处理的资源,与TC交谈以注册分支事务和报告分支事务的状态,并驱动分支事务提交或回滚

2.2. 部署seata-server(TC)

在官网下载 seata 服务端,解压后执行bin/seata-server.sh即可启动。

seata-server 有2个配置文件:registry.conf 与 file.conf。而 registry.conf 文件决定了 seata-server 应用的注册核心配置和配置信息获取形式。

咱们应用 consul 做注册核心,因而须要在registry.conf文件中,须要批改以下配置:

registry {  #file 、nacos 、eureka、redis、zk、consul、etcd3、sofa  type = "consul" ## 这里注册核心填consul  loadBalance = "RandomLoadBalance"  loadBalanceVirtualNodes = 10   ... ...  consul {    cluster = "seata-server"    serverAddr = "***注册核心地址***"    #这里的dc指的是datacenter,若consul为多数据源配置须要在申请中退出dc参数。    #dc与namespace并非是seata框架自带的,文章前面将会进一步解释    dc="bj-th"    namespace="seata-courseop"  }  ... ...}config {  # file、nacos 、apollo、zk、consul、etcd3  ## 如果启动时从注册核心获取根底配置信息,填consul  ## 否则从file.conf文件中获取  type = "consul"  consul {    serverAddr = "127.0.0.1:8500"  }... ...}

其中须要留神的是,如果须要高可用部署,seata获取配置信息的形式就必须是注册核心,此时file.conf就没用了。

(当然,须要当时把file.conf文件中的配置信息迁徙到consul中)

store {  ## store mode: file、db、redis  mode = "db"... ...  ## 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"    url = "jdbc:mysql://***线上数据库地址***/seata"    user = "******"    password = "******"    minConn = 5    maxConn = 100    ## 这里的三张表须要提前在数据库建好    globalTable = "global_table"    branchTable = "branch_table"    lockTable = "lock_table"    queryLimit = 100    maxWait = 5000  }... ...}service {  #vgroup->rgroup  vgroupMapping.tx-seata="seata-server"  default.grouplist="127.0.0.1:8091"  #degrade current not support  enableDegrade = false  #disable  disable = false  max.commit.retry.timeout = "-1"  max.rollback.retry.timeout = "-1"}

其中,global_tablebranch_tablelock_table三张表须要提前在数据库中建好。

2.3 配置client端(RM与TM)

每个应用seata框架的服务都须要引入seata组件

dependencies {    api 'com.alibaba:druid-spring-boot-starter:1.1.10'    api 'mysql:mysql-connector-java:6.0.6'    api('com.alibaba.cloud:spring-cloud-alibaba-seata:2.1.0.RELEASE') {        exclude group:'io.seata', module:'seata-all'    }    api 'com.ecwid.consul:consul-api:1.4.5'    api 'io.seata:seata-all:1.4.0'}

每个服务都同样须要配置file.conf与registry.conf文件,放在resource目录下。registry.conf与server的保持一致。在file.conf文件中,除了db配置外,还须要进行client参数的配置:

client {  rm {    asyncCommitBufferLimit = 10000    lock {      retryInterval = 10      retryTimes = 30      retryPolicyBranchRollbackOnConflict = true    }    reportRetryCount = 5    tableMetaCheckEnable = false    reportSuccessEnable = false  }  tm {    commitRetryCount = 5    rollbackRetryCount = 5  }  undo {    dataValidation = true    logSerialization = "jackson"    ## 这个undo_log也须要提前在mysql中创立    logTable = "undo_log"  }  log {    exceptionRate = 100  }}

在application.yml文件中增加seata配置:

spring:  cloud:      seata: ## 留神tx-seata须要与服务端和客户端的配置文件保持一致        tx-service-group: tx-seata

另外,还须要替换我的项目的数据源,

@Primary    @Bean("dataSource")    public DataSource druidDataSource(){        DruidDataSource druidDataSource = new DruidDataSource();        druidDataSource.setUrl(url);        druidDataSource.setUsername(username);        druidDataSource.setPassword(password);        druidDataSource.setDriverClassName(driverClassName);        return new DataSourceProxy(druidDataSource);    }

至此,client端的配置也曾经实现了。

3. 性能演示

一个分布式的全局事务,整体是两阶段提交的模型。

全局事务是由若干分支事务组成的,

分支事务要满足两阶段提交的模型要求,即须要每个分支事务都具备本人的:

  • 一阶段 prepare 行为
  • 二阶段commit 或 rollback 行为

依据两阶段行为模式的不同,咱们将分支事务划分为 Automatic (Branch) Transaction ModeTCC (Branch) Transaction Mode.

3.1 AT模式

AT 模式基于反对本地ACID事务的关系型数据库:

  • 一阶段 prepare 行为:在本地事务中,一并提交业务数据更新和相应回滚日志记录。
  • 二阶段 commit 行为:马上胜利完结,主动异步批量清理回滚日志。
  • 二阶段 rollback 行为:通过回滚日志,主动生成弥补操作,实现数据回滚
间接在须要增加全局事务的办法中加上注解@GlobalTransactional
  @SneakyThrows    @GlobalTransactional    @Transactional(rollbackFor = Exception.class)    public void buy(int id, int itemId){        // 学生成订单        Order order = orderFeignDao.create(id, itemId);        // 依据订单扣减账户余额        accountFeignDao.draw(id, order.amount);    }
留神:同@Transactional一样,@GlobalTransactional若要失效也要满足:
  • 指标函数必须为public类型
  • 同一类内办法调用时,调用指标函数的办法必须通过springBeanName.method的模式来调用,不能应用this间接调用外部办法

3.2TCC模式

TCC 模式是反对把自定义的分支事务纳入到全局事务的治理中。

  • 一阶段 prepare 行为:调用自定义的 prepare 逻辑。
  • 二阶段 commit 行为:调用自定义的 commit 逻辑。
  • 二阶段 rollback 行为:调用自定义的 rollback 逻辑。
首先编写一个TCC服务接口:
@LocalTCCpublic interface BusinessAction {    @TwoPhaseBusinessAction(name = "doBusiness", commitMethod = "commit", rollbackMethod = "rollback")    boolean doBusiness(BusinessActionContext businessActionContext,                       @BusinessActionContextParameter(paramName = "message") String msg);    boolean commit(BusinessActionContext businessActionContext);    boolean rollback(BusinessActionContext businessActionContext);}
其中,BusinessActionContext为全局事务上下文,能够从此对象中获取全局事务相干信息(如果是发动全局事务方,传入null后主动生成),而后实现该接口:
@Slf4j@Servicepublic class BusinessActionImpl implements BusinessAction {    @Transactional(rollbackFor = Exception.class)    @Override    public boolean doBusiness(BusinessActionContext businessActionContext, String msg) {        log.info("筹备do business:{}",msg);        return true;    }    @Transactional(rollbackFor = Exception.class)    @Override    public boolean commit(BusinessActionContext businessActionContext) {        log.info("business曾经commit");        return true;    }    @Transactional(rollbackFor = Exception.class)    @Override    public boolean rollback(BusinessActionContext businessActionContext) {        log.info("business曾经rollback");        return true;    }}
最初,开启全局事务办法同AT模式。
@SneakyThrows    @GlobalTransactional    public void doBusiness(BusinessActionContext context, String msg){        accountFeignDao.draw(3, new BigDecimal(100));        businessAction.doBusiness(context, msg);    }

4. 遇到的问题

4.1 client TM/RM 无奈注册到TC

在部署seata我的项目时经常会遇到这样的问题:在本地调试时一切正常,然而当试图部署到线上时,总是在clinet端提醒注册TC端失败。

  • 这是因为client须要先通过服务发现,找到注册中心里seata-server的服务信息,而后再与seata-server建设连贯。不过线上的consul采纳了多数据中心模式,在调用consul api时,必须加上dc参数项,否则将无奈返回正确的服务信息;然而,seata提供的consul服务发现组件仿佛并不反对dc参数的配置。
  • 还有一个起因也会导致client无奈连贯到TC:seata的consul客户端在调用服务状态监控api时,应用了wait与index参数,从而使consul查问进入了阻塞查问模式。此时client对consul中要查问的key做监听,只有当key发生变化或者达到最大申请工夫时,才会返回后果。貌似因为consul版本的问题,这个阻塞查问并没有监听到key的变动,反而会让服务发现的线程陷入有限期待之中,天然也就无奈让client获取到server的注册信息了。

4.2 高可用部署

seata服务的高可用部署只反对注册核心模式。因而,咱们须要想方法将file.conf文件以键值对的模式存到consul中。

遗憾的是,consul并没有显式反对namespace,咱们只能在put申请中用“/”为分隔符起到相似的成果。当然,seata框架也没有思考到这一点。所以咱们须要批改源码中的Configuration接口与RegistryProvider接口的consul实现类,减少namespace属性

4.3global_log与branch_log

TC在想mysql插入日志数据时,偶然会报:

Caused by: java.sql.SQLException: Incorrect string value:

application_data字段其实就是对业务数据的记录。官网给出的建表语句是这样的:

CREATE TABLE IF NOT EXISTS `global_table`(    `xid`                       VARCHAR(128) NOT NULL,    `transaction_id`            BIGINT,    `status`                    TINYINT      NOT NULL,    `application_id`            VARCHAR(32),    `transaction_service_group` VARCHAR(32),    `transaction_name`          VARCHAR(128),    `timeout`                   INT,    `begin_time`                BIGINT,    `application_data`          VARCHAR(2000),    `gmt_create`                DATETIME,    `gmt_modified`              DATETIME,    PRIMARY KEY (`xid`),    KEY `idx_gmt_modified_status` (`gmt_modified`, `status`),    KEY `idx_transaction_id` (`transaction_id`)) ENGINE = InnoDB  DEFAULT CHARSET = utf8;

显然,VARCHAR(2000)的大小是不适合的, utf8的格局也是不适合的。所以咱们须要批改seata对于数据源连贯的局部代码:

// connectionInitSql设置    protected Set<String> getConnectionInitSqls(){        Set<String> set = new HashSet<>();        String connectionInitSqls = CONFIG.getConfig(ConfigurationKeys.STORE_DB_CONNECTION_INIT_SQLS);        if(StringUtils.isNotEmpty(connectionInitSqls)) {            String[] strs = connectionInitSqls.split(",");            for(String s:strs){                set.add(s);            }        }        // 默认反对utf8mb4        set.add("set names utf8mb4");        return set;    }

5. 自定义开发

5.1 利用SPI机制编写自定义组件

seata基于java的spi机制提供了自定义实现接口的性能,咱们只须要在本人的服务中,依据seata的接口写好本人的实现类即可。

SPI(Service Provider Interface)是JDK内置的服务发现机制,用在不同模块间通过接口调用服务,防止对具体服务服务接口具体实现类的耦合。比方JDBC的数据库驱动模块,不同数据库连贯驱动接口雷同但实现类不同,在应用SPI机制以前调用驱动代码须要间接在类里采纳Class.forName(具体实现类全名)的形式调用,这样调用方依赖了具体的驱动实现,在替换驱动实现时要批改代码。

ConsulRegistryProvider为例:

  • ConsulRegistryServiceImpl

    // 减少DC和namespace  private static String NAMESPACE;  private static String DC;  private ConsulConfiguration() {      Config registryCongig = ConfigFactory.parseResources("registry.conf");      NAMESPACE = registryCongig.getString("config.consul.namespace");      DC = CommonSeataConfiguration.getDatacenter();      consulNotifierExecutor = new ThreadPoolExecutor(THREAD_POOL_NUM, THREAD_POOL_NUM, Integer.MAX_VALUE,              TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(),              new NamedThreadFactory("consul-config-executor", THREAD_POOL_NUM));  }  ... ...// 同时在getHealthyServices中,删除申请参数wait&index      /**   * get healthy services   *   * @param service   * @return   */  private Response<List<HealthService>> getHealthyServices(String service, long index, long watchTimeout) {      return getConsulClient().getHealthServices(service, HealthServicesRequest.newBuilder()              .setTag(SERVICE_TAG)              .setDatacenter(DC)              .setPassing(true)              .build());  }
  • ConsulRegistryProvider 留神order要大于seata包中的默认值1,seata类加载器会优先加载order更大的实现类

    @LoadLevel(name = "Consul" ,order = 2)public class ConsulRegistryProvider implements RegistryProvider { @Override public RegistryService provide() {     return ConsulRegistryServiceImpl.getInstance(); }}
  • 而后在META-INF 的services目录下增加:io.seata.discovery.registry.RegistryProvider

    com.youdao.ke.courseop.common.seata.ConsulRegistryProvider

    这样就能够替换seata包中的实现了。

5.2 common-seata工具包

对于这些自定义实现类,以及一些公共client配置,咱们能够对立封装到一个工具包下:

这样,其余我的项目只须要引入这个工具包,就能够无需繁琐的配置,间接应用了。

gradle引入common包:
api 'com.youdao.ke.courseop.common:common-seata:0.0.+'

6. 落地实例

以一个物流场景为例:
业务架构

  • logistics-server (物流服务)
  • logistics-k3c-server (物流-金蝶客户端,封装调用金蝶服务的api
  • elasticsearch

业务背景:logistics 执行领用单新增,在 elasticsearch 中更新数据,同时通过 rpc 调用 logistics-k3c 的金蝶出库办法,生成金蝶单据,如图2所示

问题:如果elasticsearch单据更新出现异常,金蝶单据将无奈回滚,造成数据不统一的问题。

在部署完seata线上服务后,只须要在logistics与logistics-k3c中别离引入common-seata工具包

logistics服务

 // 应用全局事务注解开启全局事务    @GlobalTransactional    @Transactional(rollbackFor = Exception.class)    public void Scm通过(StaffOutStockDoc staffOutStock, String body) throws Exception {        ... 一些业务解决...         // 构建金蝶单据申请        K3cApi.StaffoutstockReq req = new K3cApi.StaffoutstockReq();        req.materialNums = materialNums;        req.staffOutStockId = staffOutStock.id;        ... 一些业务解决 ...       // 调用logistics-k3c-api金蝶出库        k3cApi.staffoutstockAuditPass(req);        staffOutStock.status = 待发货;        staffOutStock.scmAuditTime = new Date();        staffOutStock.updateTime = new Date();        staffOutStock.historyPush("scm通过");        // 更新对象后存入elasticsearch        es.set(staffOutStock);    }

logistics-k3c

因为咱们新增单据接口是调用金蝶的服务,所以这里应用TCC模式构建事务接口

  • 首先创立StaffoutstockCreateAction接口

    @LocalTCCpublic interface StaffoutstockCreateAction {  @TwoPhaseBusinessAction(name = "staffoutstockCreate")  boolean create(BusinessActionContext businessActionContext,                     @BusinessActionContextParameter(paramName = "staffOutStock") StaffOutStock staffOutStock,                     @BusinessActionContextParameter(paramName = "materialNum") List<Triple<Integer, Integer, Integer>> materialNum);  boolean commit(BusinessActionContext businessActionContext);  boolean rollback(BusinessActionContext businessActionContext);}
  • 接口实现StaffoutstockCreateActionImpl

    @Slf4j@Servicepublic class StaffoutstockCreateActionImpl implements StaffoutstockCreateAction {  @Autowired  private K3cAction4Staffoutstock k3cAction4Staffoutstock;  @SneakyThrows  @Transactional(rollbackFor = Exception.class)  @Override  public boolean create(BusinessActionContext businessActionContext, StaffOutStock staffOutStock, List<Triple<Integer, Integer, Integer>> materialNum) {      //金蝶单据新增      k3cAction4Staffoutstock.staffoutstockAuditPass(staffOutStock, materialNum);      return true;  }  @SneakyThrows  @Transactional(rollbackFor = Exception.class)  @Override  public boolean commit(BusinessActionContext businessActionContext) {      Map<String, Object> context = businessActionContext.getActionContext();      JSONObject staffOutStockJson = (JSONObject) context.get("staffOutStock");      // 如果尝试新增胜利,commit不做任何事      StaffOutStock staffOutStock = staffOutStockJson.toJavaObject(StaffOutStock.class);      log.info("staffoutstock {} commit successfully!", staffOutStock.id);      return true;  }  @SneakyThrows  @Transactional(rollbackFor = Exception.class)  @Override  public boolean rollback(BusinessActionContext businessActionContext) {      Map<String, Object> context = businessActionContext.getActionContext();      JSONObject staffOutStockJson = (JSONObject) context.get("staffOutStock");      StaffOutStock staffOutStock = staffOutStockJson.toJavaObject(StaffOutStock.class);      // 这里调用金蝶单据删除接口进行回滚      k3cAction4Staffoutstock.staffoutstockRollback(staffOutStock);      log.info("staffoutstock {} rollback successfully!", staffOutStock.id);      return true;  }}
  • 封装为业务办法

    /**   * 项目组领用&报废的审核通过:新增其余出库单   * 该办法应用seata-TCC计划实现全局事务   * @param staffOutStock   * @param materialNum   */    @Transactional  public void staffoutstockAuditPassWithTranscation(StaffOutStock staffOutStock,                                                    List<Triple<Integer, Integer, Integer>> materialNum){      staffoutstockCreateAction.create(null, staffOutStock, materialNum);  }
  • k3c API实现类

     @SneakyThrows  @Override  public void staffoutstockAuditPass(StaffoutstockReq req) {      ... 一些业务解决办法 ...      //这里调用了封装好的事务办法      k3cAction4Staffoutstock.staffoutstockAuditPassWithTranscation(staffOutStock, triples);  }

这样,一个基于 TCC 的全局事务链路就建设起来了。

当全局事务执行胜利时,咱们能够在 server 中看到打印的日志(如图3):

如果全局事务执行失败,会进行回滚,此时会执行接口中的rollback,调用金蝶接口删除生成的单据,如图4。

7. 总结

本文以seata框架的部署与应用为主线,记录了seata 框架使用的一些关键步骤与技术细节,并针对我的项目落地时遇到的一些的技术问题提供了解决方案。

在后续的推文中,咱们还将持续以 seata 框架的源码解析为主线,向大家介绍 seata 实现分布式事务的外围原理与技术细节。
-END-