Spring Boot集成Seata实现基于AT模式的分布式事务的解决方案_Java

1.什么是seata？

seata 是一款开源的分布式事务解决方案，致力于提供高性能和简单易用的分布式事务服务。seata 将为用户提供了 at、tcc、saga 和 xa 事务模式，为用户打造一站式的分布式解决方案。

at 模式

前提

基于支持本地 acid 事务的关系型数据库。
java 应用，通过 jdbc 访问数据库。

整体机制

两阶段提交协议的演变：

一阶段：业务数据和回滚日志记录在同一个本地事务中提交，释放本地锁和连接资源。
二阶段：
- 提交异步化，非常快速地完成。
- 回滚通过一阶段的回滚日志进行反向补偿。

写隔离

一阶段本地事务提交前，需要确保先拿到 全局锁 。
拿不到 全局锁 ，不能提交本地事务。
拿 全局锁 的尝试被限制在一定范围内，超出范围将放弃，并回滚本地事务，释放本地锁。

以一个示例来说明：两个全局事务 tx1 和 tx2，分别对 a 表的 m 字段进行更新操作，m 的初始值 1000。 tx1 先开始，开启本地事务，拿到本地锁，更新操作 m = 1000 - 100 = 900。本地事务提交前，先拿到该记录的 全局锁 ，本地提交释放本地锁。 tx2 后开始，开启本地事务，拿到本地锁，更新操作 m = 900 - 100 = 800。本地事务提交前，尝试拿该记录的 全局锁 ，tx1 全局提交前，该记录的全局锁被 tx1 持有，tx2 需要重试等待 全局锁 。

tx1 二阶段全局提交，释放 全局锁 。tx2 拿到 全局锁 提交本地事务。

如果 tx1 的二阶段全局回滚，则 tx1 需要重新获取该数据的本地锁，进行反向补偿的更新操作，实现分支的回滚。此时，如果 tx2 仍在等待该数据的 全局锁，同时持有本地锁，则 tx1 的分支回滚会失败。分支的回滚会一直重试，直到 tx2 的 全局锁 等锁超时，放弃 全局锁 并回滚本地事务释放本地锁，tx1 的分支回滚最终成功。因为整个过程 全局锁 在 tx1 结束前一直是被 tx1 持有的，所以不会发生脏写的问题。

读隔离

在数据库本地事务隔离级别 读已提交（read committed） 或以上的基础上，seata（at 模式）的默认全局隔离级别是 读未提交（read uncommitted） 。如果应用在特定场景下，必需要求全局的 读已提交 ，目前 seata 的方式是通过 select for update 语句的代理。

read isolation: select for update

select for update 语句的执行会申请 全局锁 ，如果 全局锁 被其他事务持有，则释放本地锁（回滚 select for update 语句的本地执行）并重试。这个过程中，查询是被 block 住的，直到 全局锁 拿到，即读取的相关数据是 已提交 的，才返回。

出于总体性能上的考虑，seata 目前的方案并没有对所有 select 语句都进行代理，仅针对 for update 的 select 语句。

具体例子相见：what is seata? | apache seata

2.环境搭建

安装mysql

参见代码仓库里面的mysql模块里面的docker文件夹

install seta-server

version: "3.1"
services:
  seata-server:
    image: seataio/seata-server:latest
    hostname: seata-server
    ports:
      - "7091:7091"
      - "8091:8091"
    environment:
      - seata_port=8091
      - store_mode=file

http://localhost:7091/#/overview

default username and password is admin/admin

3.代码工程

实验目标

订单服务调用库存服务和账户余额服务进行相应的扣减，并且最终生成订单

seata-order

订单服务

pom.xml

<?xml version="1.0" encoding="utf-8"?>
<project xmlns="http://maven.apache.org/pom/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/xmlschema-instance"
         xsi:schemalocation="http://maven.apache.org/pom/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactid>seata</artifactid>
        <groupid>com.et</groupid>
        <version>1.0-snapshot</version>
    </parent>
    <modelversion>4.0.0</modelversion>
    <artifactid>seata-order</artifactid>
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
    <dependencies>
        <dependency>
            <groupid>org.springframework.boot</groupid>
            <artifactid>spring-boot-starter-web</artifactid>
        </dependency>
        <dependency>
            <groupid>org.springframework.boot</groupid>
            <artifactid>spring-boot-autoconfigure</artifactid>
        </dependency>
        <dependency>
            <groupid>org.springframework.boot</groupid>
            <artifactid>spring-boot-starter-test</artifactid>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupid>org.springframework.boot</groupid>
            <artifactid>spring-boot-starter-jdbc</artifactid>
        </dependency>
        <dependency>
            <groupid>mysql</groupid>
            <artifactid>mysql-connector-java</artifactid>
            <version>5.1.48</version>
        </dependency>
        <dependency>
            <groupid>org.mybatis.spring.boot</groupid>
            <artifactid>mybatis-spring-boot-starter</artifactid>
            <version>2.1.2</version>
        </dependency>
        <dependency>
            <groupid>io.seata</groupid>
            <artifactid>seata-spring-boot-starter</artifactid>
            <version>1.1.0</version>
        </dependency>
        <dependency>
            <groupid>io.seata</groupid>
            <artifactid>seata-http</artifactid>
            <version>1.1.0</version>
        </dependency>
        <dependency>
            <groupid>org.apache.httpcomponents</groupid>
            <artifactid>httpclient</artifactid>
            <version>4.5.8</version>
        </dependency>
        <dependency>
            <groupid>org.projectlombok</groupid>
            <artifactid>lombok</artifactid>
        </dependency>
    </dependencies>
</project>

controller

package com.et.seata.order.controller;
import com.et.seata.order.service.orderservice;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.web.bind.annotation.postmapping;
import org.springframework.web.bind.annotation.requestmapping;
import org.springframework.web.bind.annotation.requestparam;
import org.springframework.web.bind.annotation.restcontroller;
import java.io.ioexception;
import java.util.hashmap;
import java.util.map;
@restcontroller
public class helloworldcontroller {
    @autowired
    private orderservice orderservice;
    @postmapping("/create")
    public map<string, object> createorder(@requestparam("userid") long userid,
                                              @requestparam("productid") long productid,
                                              @requestparam("price") integer price) throws ioexception {
        map<string, object> map = new hashmap<>();
        map.put("msg", "helloworld");
        map.put("reuslt", orderservice.createorder(userid,productid,price));
        return map;
    }
}

service

package com.et.seata.order.service;
import com.alibaba.fastjson.jsonobject;
import com.et.seata.order.dao.orderdao;
import com.et.seata.order.dto.orderdo;
import io.seata.core.context.rootcontext;
import io.seata.integration.http.defaulthttpexecutor;
import io.seata.spring.annotation.globaltransactional;
import lombok.extern.slf4j.slf4j;
import org.apache.http.httpresponse;
import org.apache.http.util.entityutils;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.stereotype.service;
import java.io.ioexception;
/**
 * @author liuhaihua
 * @version 1.0
 * @classname orderserviceimpl
 * @description todo
 * @date 2024/08/08/ 13:53
 */
@slf4j
@service
public class orderserviceimpl implements orderservice{
   @autowired
   orderdao orderdao;
    @override
    @globaltransactional // <1>
    public integer createorder(long userid, long productid, integer price) throws ioexception {
        integer amount = 1; // 购买数量，暂时设置为 1。
        log.info("[createorder] 当前 xid: {}", rootcontext.getxid());
        // <2> 扣减库存
        this.reducestock(productid, amount);
        // <3> 扣减余额
        this.reducebalance(userid, price);
        // <4> 保存订单
        log.info("[createorder] 保存订单");
        return this.saveorder(userid,productid,price,amount);
    }
    private integer saveorder(long userid, long productid, integer price,integer amount){
      // <4> 保存订单
      orderdo order = new orderdo();
      order.setuserid(userid);
      order.setproductid(productid);
      order.setpayamount(amount * price);
      orderdao.saveorder(order);
      log.info("[createorder] 保存订单: {}", order.getid());
      return order.getid();
   }
    private void reducestock(long productid, integer amount) throws ioexception {
        // 参数拼接
        jsonobject params = new jsonobject().fluentput("productid", string.valueof(productid))
                .fluentput("amount", string.valueof(amount));
        // 执行调用
        httpresponse response = defaulthttpexecutor.getinstance().executepost("http://127.0.0.1:8082", "/stock",
                params, httpresponse.class);
        // 解析结果
        boolean success = boolean.valueof(entityutils.tostring(response.getentity()));
        if (!success) {
            throw new runtimeexception("扣除库存失败");
        }
    }
    private void reducebalance(long userid, integer price) throws ioexception {
        // 参数拼接
        jsonobject params = new jsonobject().fluentput("userid", string.valueof(userid))
                .fluentput("price", string.valueof(price));
        // 执行调用
        httpresponse response = defaulthttpexecutor.getinstance().executepost("http://127.0.0.1:8083", "/balance",
                params, httpresponse.class);
        // 解析结果
        boolean success = boolean.valueof(entityutils.tostring(response.getentity()));
        if (!success) {
            throw new runtimeexception("扣除余额失败");
        }
    }
}

application.yaml

server:
  port: 8081 # 端口
spring:
  application:
    name: order-service
  datasource:
    url: jdbc:mysql://127.0.0.1:3306/seata_order?usessl=false&useunicode=true&characterencoding=utf-8
    driver-class-name: com.mysql.jdbc.driver
    username: root
    password: 123456
# seata 配置项，对应 seataproperties 类
seata:
  application-id: ${spring.application.name} # seata 应用编号，默认为 ${spring.application.name}
  tx-service-group: ${spring.application.name}-group # seata 事务组编号，用于 tc 集群名
  # 服务配置项，对应 serviceproperties 类
  service:
    # 虚拟组和分组的映射
    vgroup-mapping:
      order-service-group: default
    # 分组和 seata 服务的映射
    grouplist:
      default: 127.0.0.1:8091

seata-product

商品库存服务

controller

package com.et.seata.product.controller;
import com.et.seata.product.dto.productreducestockdto;
import com.et.seata.product.service.productservice;
import lombok.extern.slf4j.slf4j;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.web.bind.annotation.postmapping;
import org.springframework.web.bind.annotation.requestbody;
import org.springframework.web.bind.annotation.restcontroller;
@restcontroller
@slf4j
public class productcontroller {
   @autowired
   productservice productservice;
   @postmapping("/stock")
   public boolean reducestock(@requestbody productreducestockdto productreducestockdto) {
      log.info("[reducestock] 收到减少库存请求, 商品:{}, 价格:{}", productreducestockdto.getproductid(),
            productreducestockdto.getamount());
      try {
         productservice.reducestock(productreducestockdto.getproductid(), productreducestockdto.getamount());
         // 正常扣除库存，返回 true
         return true;
      } catch (exception e) {
         // 失败扣除库存，返回 false
         return false;
      }
   }
}

service

package com.et.seata.product.service;
import com.et.seata.product.dao.productdao;
import io.seata.core.context.rootcontext;
import lombok.extern.slf4j.slf4j;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.stereotype.service;
import org.springframework.transaction.annotation.transactional;
@service
@slf4j
public class productserviceimpl implements productservice {
    @autowired
    private productdao productdao;
    @override
    @transactional // <1> 开启新事物
    public void reducestock(long productid, integer amount) throws exception {
        log.info("[reducestock] 当前 xid: {}", rootcontext.getxid());
        // <2> 检查库存
        checkstock(productid, amount);
        log.info("[reducestock] 开始扣减 {} 库存", productid);
        // <3> 扣减库存
        int updatecount = productdao.reducestock(productid, amount);
        // 扣除成功
        if (updatecount == 0) {
            log.warn("[reducestock] 扣除 {} 库存失败", productid);
            throw new exception("库存不足");
        }
        // 扣除失败
        log.info("[reducestock] 扣除 {} 库存成功", productid);
    }
    private void checkstock(long productid, integer requiredamount) throws exception {
        log.info("[checkstock] 检查 {} 库存", productid);
        integer stock = productdao.getstock(productid);
        if (stock < requiredamount) {
            log.warn("[checkstock] {} 库存不足，当前库存: {}", productid, stock);
            throw new exception("库存不足");
        }
    }
}

dao

package com.et.seata.product.dao;
import org.apache.ibatis.annotations.mapper;
import org.apache.ibatis.annotations.param;
import org.apache.ibatis.annotations.select;
import org.apache.ibatis.annotations.update;
import org.springframework.stereotype.repository;
@mapper
@repository
public interface productdao {
    /**
     * 获取库存
     *
     * @param productid 商品编号
     * @return 库存
     */
    @select("select stock from product where id = #{productid}")
    integer getstock(@param("productid") long productid);
    /**
     * 扣减库存
     *
     * @param productid 商品编号
     * @param amount    扣减数量
     * @return 影响记录行数
     */
    @update("update product set stock = stock - #{amount} where id = #{productid} and stock >= #{amount}")
    int reducestock(@param("productid") long productid, @param("amount") integer amount);
}

seata-balance

用户余额服务

controller

package com.et.seata.balance.controller;
import com.et.seata.balance.dto.accountreducebalancedto;
import com.et.seata.balance.service.accountservice;
import lombok.extern.slf4j.slf4j;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.web.bind.annotation.postmapping;
import org.springframework.web.bind.annotation.requestbody;
import org.springframework.web.bind.annotation.requestmapping;
import org.springframework.web.bind.annotation.restcontroller;
import java.util.hashmap;
import java.util.map;
@restcontroller
@slf4j
public class accountcontroller {
   @autowired
   private accountservice accountservice;
   @postmapping("/balance")
   public boolean reducebalance(@requestbody accountreducebalancedto accountreducebalancedto) {
      log.info("[reducebalance] 收到减少余额请求, 用户:{}, 金额:{}", accountreducebalancedto.getuserid(),
            accountreducebalancedto.getprice());
      try {
         accountservice.reducebalance(accountreducebalancedto.getuserid(), accountreducebalancedto.getprice());
         // 正常扣除余额，返回 true
         return true;
      } catch (exception e) {
         // 失败扣除余额，返回 false
         return false;
      }
   }
}

service

package com.et.seata.balance.service;
import com.et.seata.balance.dao.accountdao;
import io.seata.core.context.rootcontext;
import lombok.extern.slf4j.slf4j;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.stereotype.service;
import org.springframework.transaction.annotation.propagation;
import org.springframework.transaction.annotation.transactional;
@service
@slf4j
public class accountserviceimpl implements accountservice {
    @autowired
    private accountdao accountdao;
    @override
    @transactional(propagation = propagation.requires_new) // <1> 开启新事物
    public void reducebalance(long userid, integer price) throws exception {
        log.info("[reducebalance] 当前 xid: {}", rootcontext.getxid());
        // <2> 检查余额
        checkbalance(userid, price);
        log.info("[reducebalance] 开始扣减用户 {} 余额", userid);
        // <3> 扣除余额
        int updatecount = accountdao.reducebalance(price);
        // 扣除成功
        if (updatecount == 0) {
            log.warn("[reducebalance] 扣除用户 {} 余额失败", userid);
            throw new exception("余额不足");
        }
        log.info("[reducebalance] 扣除用户 {} 余额成功", userid);
    }
    private void checkbalance(long userid, integer price) throws exception {
        log.info("[checkbalance] 检查用户 {} 余额", userid);
        integer balance = accountdao.getbalance(userid);
        if (balance < price) {
            log.warn("[checkbalance] 用户 {} 余额不足，当前余额:{}", userid, balance);
            throw new exception("余额不足");
        }
    }
}

dao

package com.et.seata.balance.dao;
import org.apache.ibatis.annotations.mapper;
import org.apache.ibatis.annotations.param;
import org.apache.ibatis.annotations.select;
import org.apache.ibatis.annotations.update;
import org.springframework.stereotype.repository;
@mapper
@repository
public interface accountdao {
    /**
     * 获取账户余额
     *
     * @param userid 用户 id
     * @return 账户余额
     */
    @select("select balance from account where id = #{userid}")
    integer getbalance(@param("userid") long userid);
    /**
     * 扣减余额
     *
     * @param price 需要扣减的数目
     * @return 影响记录行数
     */
    @update("update account set balance = balance - #{price} where id = 1 and balance >= ${price}")
    int reducebalance(@param("price") integer price);
}

以上只是一些关键代码，所有代码请参见下面代码仓库

代码仓库

https://github.com/harries/springboot-demo

4.测试

启动seata-order服务
启动seata-product服务
启动seata-balance服务

编辑可以看到控制台输出回滚日志

2024-08-08 22:00:59.467 info 35051 --- [tch_rmrole_1_16] i.s.core.rpc.netty.rmmessagelistener : onmessage:xid=172.22.0.3:8091:27573281007513609,branchid=27573281007513610,branchtype=at,resourceid=jdbc:mysql://127.0.0.1:3306/seata_storage,applicationdata=null
2024-08-08 22:00:59.467 info 35051 --- [tch_rmrole_1_16] io.seata.rm.abstractrmhandler : branch rollbacking: 172.22.0.3:8091:27573281007513609 27573281007513610 jdbc:mysql://127.0.0.1:3306/seata_storage
2024-08-08 22:00:59.503 info 35051 --- [tch_rmrole_1_16] i.s.r.d.undo.abstractundologmanager : xid 172.22.0.3:8091:27573281007513609 branch 27573281007513610, undo_log deleted with globalfinished
2024-08-08 22:00:59.511 info 35051 --- [tch_rmrole_1_16] io.seata.rm.abstractrmhandler : branch rollbacked result: phasetwo_rollbacked