Spring Boot 集成 Seata 解决分布式事务问题

文章首发于公众号《程序员果果》
地址 : https://mp.weixin.qq.com/s/aDhGG3Y2t4lPYetK01Wmxg

seata 简介

Seata 是 阿里巴巴2019年开源的分布式事务解决方案，致力于在微服务架构下提供高性能和简单易用的分布式事务服务。在 Seata 开源之前，Seata 对应的内部版本在阿里内部一直扮演着分布式一致性中间件的角色，帮助阿里度过历年的双11，对各业务进行了有力的支撑。经过多年沉淀与积累，2019.1 Seata 正式宣布对外开源 。目前 Seata 1.0 已经 GA。

微服务中的分布式事务问题

让我们想象一下传统的单片应用程序，它的业务由3个模块组成，他们使用单个本地数据源。自然，本地事务将保证数据的一致性。

微服务架构已发生了变化。上面提到的3个模块被设计为3种服务。本地事务自然可以保证每个服务中的数据一致性。但是整个业务逻辑范围如何？

Seata怎么办？

我们说，分布式事务是由一批分支事务组成的全局事务，通常分支事务只是本地事务。

Seata有3个基本组成部分：

  • 事务协调器（TC）：维护全局事务和分支事务的状态，驱动全局提交或回滚。
  • 事务管理器TM：定义全局事务的范围：开始全局事务，提交或回滚全局事务。
  • 资源管理器（RM）：管理正在处理的分支事务的资源，与TC对话以注册分支事务并报告分支事务的状态，并驱动分支事务的提交或回滚。

Seata管理的分布式事务的典型生命周期：

  1. TM要求TC开始一项新的全局事务。TC生成代表全局事务的XID。
  2. XID通过微服务的调用链传播。
  3. RM将本地事务注册为XID到TC的相应全局事务的分支。
  4. TM要求TC提交或回退相应的XID全局事务。
  5. TC驱动XID的相应全局事务下的所有分支事务以完成分支提交或回滚。

快速开始

用例

用户购买商品的业务逻辑。整个业务逻辑由3个微服务提供支持：

  • 仓储服务：对给定的商品扣除仓储数量。
  • 订单服务：根据采购需求创建订单。
  • 帐户服务：从用户帐户中扣除余额。

环境准备

步骤 1：建立数据库

# db_seata
DROP SCHEMA IF EXISTS db_seata;
CREATE SCHEMA db_seata;
USE db_seata;

# Account
CREATE TABLE `account_tbl` (
  `id` INT(11) NOT NULL AUTO_INCREMENT,
  `user_id` VARCHAR(255) DEFAULT NULL,
  `money` INT(11) DEFAULT 0,
  PRIMARY KEY (`id`)
) ENGINE = InnoDB DEFAULT CHARSET = utf8;

INSERT INTO account_tbl (id, user_id, money)
VALUES (1, ‘1001‘, 10000);
INSERT INTO account_tbl (id, user_id, money)
VALUES (2, ‘1002‘, 10000);

# Order
CREATE TABLE `order_tbl`
(
  `id` INT(11) NOT NULL AUTO_INCREMENT,
  `user_id` VARCHAR(255) DEFAULT NULL,
  `commodity_code` VARCHAR(255) DEFAULT NULL,
  `count` INT(11) DEFAULT ‘0‘,
  `money` INT(11) DEFAULT ‘0‘,
  PRIMARY KEY (`id`)
) ENGINE = InnoDB DEFAULT CHARSET = utf8;

# Storage
CREATE TABLE `storage_tbl` (
  `id` INT(11) NOT NULL AUTO_INCREMENT,
  `commodity_code` VARCHAR(255) DEFAULT NULL,
  `count` INT(11) DEFAULT ‘0‘,
  PRIMARY KEY (`id`),
  UNIQUE KEY `commodity_code` (`commodity_code`)
) ENGINE = InnoDB DEFAULT CHARSET = utf8;

INSERT INTO storage_tbl (id, commodity_code, count)
VALUES (1, ‘2001‘, 1000);

CREATE TABLE `undo_log` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `branch_id` bigint(20) NOT NULL,
  `xid` varchar(100) NOT NULL,
  `context` varchar(128) NOT NULL,
  `rollback_info` longblob NOT NULL,
  `log_status` int(11) NOT NULL,
  `log_created` datetime NOT NULL,
  `log_modified` datetime NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

seata AT 模式需要 undo_log 表,另外三张是业务表。

步骤 2: 启动 Seata Server

Server端存储模式（store.mode）现有file、db两种（后续将引入raft），file模式无需改动，直接启动即可。db模式需要导入用于存储全局事务回话信息的三张表。

注：file模式为单机模式，全局事务会话信息内存中读写并持久化本地文件root.data，性能较高;
db模式为高可用模式，全局事务会话信息通过db共享，相应性能差些

可以直接通过bash 脚本启动 Seata Server，也可以通过 Docker 镜像启动，但是 Docker 方式目前只支持使用 file 模式，不支持将 Seata-Server 注册到 Eureka 或 Nacos 等注册中心。

通过脚本启动

在 https://github.com/seata/seata/releases 下载相应版本的 Seata Server，解压后执行以下命令启动，这里使用 file 配置

通过 Docker 启动
docker run --name seata-server -p 8091:8091 seataio/seata-server:latest

项目介绍

项目名 地址 说明
sbm-account-service 127.0.0.1:8081 账户服务
sbm-order-service 127.0.0.1:8082 订单服务
sbm-storage-service 127.0.0.1:8083 仓储服务
sbm-business-service 127.0.0.1:8084 主业务
seata-server 172.16.2.101:8091 seata-server

核心代码

为了不让篇幅太长，这里只给出部分代码，详细代码文末会给出源码地址

maven 引入 seata 的依赖 eata-spring-boot-starter

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

仓储服务

application.properties
spring.application.name=account-service
server.port=8081
spring.datasource.url=jdbc:mysql://172.16.2.101:3306/db_seata?useSSL=false&serverTimezone=UTC
spring.datasource.username=root
spring.datasource.password=123456
seata.tx-service-group=my_test_tx_group
mybatis.mapper-locations=classpath*:mapper/*Mapper.xml
seata.service.grouplist=172.16.2.101:8091
logging.level.io.seata=info
logging.level.io.seata.samples.account.persistence.AccountMapper=debug
StorageService
public interface StorageService {

    /**
     * 扣除存储数量
     */
    void deduct(String commodityCode, int count);
}

订单服务

public interface OrderService {

    /**
     * 创建订单
     */
    Order create(String userId, String commodityCode, int orderCount);
}

帐户服务

public interface AccountService {

    /**
     * 从用户账户中借出
     */
    void debit(String userId, int money);
}

主要业务逻辑

只需要使用一个 @GlobalTransactional 注解在业务方法上。

@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);
}

XID 的传递

全局事务ID的跨服务传递，需要我们自己实现，这里通过拦截器的方式。每个服务都需要添加下面两个类。

SeataFilter
@Component
public class SeataFilter implements Filter {
    @Override
    public void init(FilterConfig filterConfig) throws ServletException {
    }

    @Override
    public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws IOException, ServletException {
        HttpServletRequest req = (HttpServletRequest) servletRequest;
        String xid = req.getHeader(RootContext.KEY_XID.toLowerCase());
        boolean isBind = false;
        if (StringUtils.isNotBlank(xid)) {
            RootContext.bind(xid);
            isBind = true;
        }
        try {
            filterChain.doFilter(servletRequest, servletResponse);
        } finally {
            if (isBind) {
                RootContext.unbind();
            }
        }
    }

    @Override
    public void destroy() {
    }
}
SeataRestTemplateAutoConfiguration
@Configuration
public class SeataRestTemplateAutoConfiguration {
    @Autowired(
            required = false
    )
    private Collection<RestTemplate> restTemplates;
    @Autowired
    private SeataRestTemplateInterceptor seataRestTemplateInterceptor;

    public SeataRestTemplateAutoConfiguration() {
    }

    @Bean
    public SeataRestTemplateInterceptor seataRestTemplateInterceptor() {
        return new SeataRestTemplateInterceptor();
    }

    @PostConstruct
    public void init() {
        if (this.restTemplates != null) {
            Iterator var1 = this.restTemplates.iterator();

            while (var1.hasNext()) {
                RestTemplate restTemplate = (RestTemplate) var1.next();
                List<ClientHttpRequestInterceptor> interceptors = new ArrayList(restTemplate.getInterceptors());
                interceptors.add(this.seataRestTemplateInterceptor);
                restTemplate.setInterceptors(interceptors);
            }
        }

    }
}

测试

测试成功场景:

curl -X POST http://127.0.0.1:8084/api/business/purchase/commit

此时返回结果为：true

测试失败场景:

UserId 为1002 的用户下单，sbm-account-service会抛出异常，事务会回滚

http://127.0.0.1:8084/api/business/purchase/rollback

此时返回结果为：false

查看 undo_log 的日志或者主键，可以看到在执行过程中有保存数据。
如查看主键自增的值，在执行前后的值会发生变化，在执行前是 1，执行后是 7 。

源码地址

https://github.com/gf-huanchupk/SpringBootLearning/tree/master/springboot-seata

参考

http://seata.io/zh-cn/docs/overview/what-is-seata.html

原文地址：https://blog.51cto.com/13698036/2467230

时间： 2024-07-30 14:32:37

Spring Boot 集成 Seata 解决分布式事务问题的相关文章

Spring Boot微服务如何集成fescar解决分布式事务？

什么是fescar? 关于fescar的详细介绍,请参阅fescar wiki. 传统的2PC提交协议,会持有一个全局性的锁,所有局部事务预提交成功后一起提交,或有一个局部事务预提交失败后一起回滚,最后释放全局锁.锁持有的时间较长,会对并发造成较大的影响,死锁的风险也较高. fescar的创新之处在于,每个局部事务执行完立即提交,释放本地锁:它会去解析你代码中的sql,从数据库中获得事务提交前的事务资源即数据,存放到undo_log中,全局事务协调器在回滚的时候直接使用undo_log中的数据覆

Spring boot集成Swagger2,解决页面不显示的问题

maven <!-- swagger --><dependency> <groupId>io.springfox</groupId> <artifactId>springfox-swagger2</artifactId> <version>2.6.1</version></dependency><dependency> <groupId>io.springfox</gr

Spring Boot集成Hazelcast实现集群与分布式内存缓存

Hazelcast是Hazelcast公司开源的一款分布式内存数据库产品,提供弹性可扩展.高性能的分布式内存计算.并通过提供诸如Map,Queue,ExecutorService,Lock和JCache等Java的许多开发人员友好的分布式实现. 了解Hazelcast Hazelcast特性 简单易用 Hazelcast是用Java编写的,没有其他依赖关系.只需简单的把jar包引入项目的classpath即可创建集群. 无主从模式 与许多NoSQL解决方案不同,Hazelcast节点是点对点的.

Spring Cloud Alibaba | 微服务分布式事务之Seata

Spring Cloud Alibaba | 微服务分布式事务之Seata 本篇实战所使用Spring有关版本: SpringBoot:2.1.7.RELEASE Spring Cloud:Greenwich.SR2 Spring CLoud Alibaba:2.1.0.RELEASE 1. 概述 在构建微服务的过程中,不管是使用什么框架.组件来构建,都绕不开一个问题,跨服务的业务操作如何保持数据一致性. 2. 什么是分布式事务? 首先,设想一个传统的单体应用,无论多少内部调用,最后终归是在同一

解决Spring Boot集成Shiro，配置类使用Autowired无法注入Bean问题

如题,最近使用spring boot集成shiro,在shiroFilter要使用数据库动态给URL赋权限的时候,发现 @Autowired 注入的bean都是null,无法注入mapper.搜了半天似乎网上都没有相关问题,也是奇怪.最后发现 /** * Shiro生命周期处理器 * * @return */ @Bean(name = "lifecycleBeanPostProcessor") public LifecycleBeanPostProcessor getLifecycle

RabbitMQ解决分布式事务

案例:经典案例,以目前流行点外卖的案例,用户下单后,调用订单服务,让后订单服务调用派单系统通知送外卖人员送单,这时候订单系统与派单系统采用MQ异步通讯. RabbitMQ解决分布式事务原理: 采用最终一致性原理.需要保证以下三要素1.确认生产者一定要将数据投递到MQ服务器中(采用MQ消息确认机制)2.MQ消费者消息能够正确消费消息,采用手动ACK模式,使用不补偿机制(注意重试幂等性问题)3.如何保证第一个事务先执行,采用补偿机制(补单机制),在创建一个补单消费者进行监听,如果订单没有创建成功,进

使用kafka消息队列解决分布式事务

微服务框架Spring Cloud介绍 Part1: 使用事件和消息队列实现分布式事务 本文转自:http://skaka.me/blog/2016/04/21/springcloud1/ 不同于单一架构应用(Monolith), 分布式环境下, 进行事务操作将变得困难, 因为分布式环境通常会有多个数据源, 只用本地数据库事务难以保证多个数据源数据的一致性. 这种情况下, 可以使用两阶段或者三阶段提交协议来完成分布式事务.但是使用这种方式一般来说性能较差, 因为事务管理器需要在多个数据源之间进行

spring boot 集成 hbase

spring boot 集成 hbase 会启动报错 主要因为Spring Boot内嵌了Web容器,方便对应用进行微服务化开发和部署.所以打算将HBase的业务应用作为一个单服务进行开发和发布,其他相关的子系统通过RESTful API来访问. 搭建项目环境时,需要注意的事项: 由于Spring Boot内嵌了Web容器,所以框架默认导入了依赖:tomcat-embed-core-8.5.5.jar.tomcat-embed-el-8.5.5.jar等包.而HBase的jar中包含了:serv

spring boot集成swagger2

做java Web的后端开发已经两年多了,一般都是开发完了接口,都把接口更新到wiki文档上,然后通知前端去文档上去查阅接口的详细描述, 当时经常接口会有变动,加参数或返回值夹字段,所以维护语线上一致的文档是一件非常麻烦的事情,前一段时间同事聊天说他们公司用的swagger2,这个不需要写文档,它是自动生成文档,只要会使用它提供的几个的注解就行,于是上网找了下资料,发现它于spring boot集成也非常方便.不废话直接看了代码. 首先,在maven项目的pom.xml加上他需要的依赖. <de