Spring Boot集成Seata實(shí)現(xiàn)基于AT模式的分布式事務(wù)的解決方案
1.什么是Seata?
Seata 是一款開(kāi)源的分布式事務(wù)解決方案,致力于提供高性能和簡(jiǎn)單易用的分布式事務(wù)服務(wù)。Seata 將為用戶提供了 AT、TCC、SAGA 和 XA 事務(wù)模式,為用戶打造一站式的分布式解決方案。
AT 模式
前提?
- 基于支持本地 ACID 事務(wù)的關(guān)系型數(shù)據(jù)庫(kù)。
- Java 應(yīng)用,通過(guò) JDBC 訪問(wèn)數(shù)據(jù)庫(kù)。
整體機(jī)制?
兩階段提交協(xié)議的演變:
- 一階段:業(yè)務(wù)數(shù)據(jù)和回滾日志記錄在同一個(gè)本地事務(wù)中提交,釋放本地鎖和連接資源。
- 二階段:
- 提交異步化,非常快速地完成。
- 回滾通過(guò)一階段的回滾日志進(jìn)行反向補(bǔ)償。
寫(xiě)隔離
- 一階段本地事務(wù)提交前,需要確保先拿到 全局鎖 。
- 拿不到 全局鎖 ,不能提交本地事務(wù)。
- 拿 全局鎖 的嘗試被限制在一定范圍內(nèi),超出范圍將放棄,并回滾本地事務(wù),釋放本地鎖。
以一個(gè)示例來(lái)說(shuō)明: 兩個(gè)全局事務(wù) tx1 和 tx2,分別對(duì) a 表的 m 字段進(jìn)行更新操作,m 的初始值 1000。 tx1 先開(kāi)始,開(kāi)啟本地事務(wù),拿到本地鎖,更新操作 m = 1000 - 100 = 900。本地事務(wù)提交前,先拿到該記錄的 全局鎖 ,本地提交釋放本地鎖。 tx2 后開(kāi)始,開(kāi)啟本地事務(wù),拿到本地鎖,更新操作 m = 900 - 100 = 800。本地事務(wù)提交前,嘗試拿該記錄的 全局鎖 ,tx1 全局提交前,該記錄的全局鎖被 tx1 持有,tx2 需要重試等待 全局鎖 。

tx1 二階段全局提交,釋放 全局鎖 。tx2 拿到 全局鎖 提交本地事務(wù)。

如果 tx1 的二階段全局回滾,則 tx1 需要重新獲取該數(shù)據(jù)的本地鎖,進(jìn)行反向補(bǔ)償?shù)母虏僮?,?shí)現(xiàn)分支的回滾。 此時(shí),如果 tx2 仍在等待該數(shù)據(jù)的 全局鎖,同時(shí)持有本地鎖,則 tx1 的分支回滾會(huì)失敗。分支的回滾會(huì)一直重試,直到 tx2 的 全局鎖 等鎖超時(shí),放棄 全局鎖 并回滾本地事務(wù)釋放本地鎖,tx1 的分支回滾最終成功。 因?yàn)檎麄€(gè)過(guò)程 全局鎖 在 tx1 結(jié)束前一直是被 tx1 持有的,所以不會(huì)發(fā)生 臟寫(xiě) 的問(wèn)題。
讀隔離
在數(shù)據(jù)庫(kù)本地事務(wù)隔離級(jí)別 讀已提交(Read Committed) 或以上的基礎(chǔ)上,Seata(AT 模式)的默認(rèn)全局隔離級(jí)別是 讀未提交(Read Uncommitted) 。 如果應(yīng)用在特定場(chǎng)景下,必需要求全局的 讀已提交 ,目前 Seata 的方式是通過(guò) SELECT FOR UPDATE 語(yǔ)句的代理。

SELECT FOR UPDATE 語(yǔ)句的執(zhí)行會(huì)申請(qǐng) 全局鎖 ,如果 全局鎖 被其他事務(wù)持有,則釋放本地鎖(回滾 SELECT FOR UPDATE 語(yǔ)句的本地執(zhí)行)并重試。這個(gè)過(guò)程中,查詢是被 block 住的,直到 全局鎖 拿到,即讀取的相關(guān)數(shù)據(jù)是 已提交 的,才返回。
出于總體性能上的考慮,Seata 目前的方案并沒(méi)有對(duì)所有 SELECT 語(yǔ)句都進(jìn)行代理,僅針對(duì) FOR UPDATE 的 SELECT 語(yǔ)句。
具體例子相見(jiàn):What Is Seata? | Apache Seata
2.環(huán)境搭建
安裝mysql
參見(jiàn)代碼倉(cāng)庫(kù)里面的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=filehttp://localhost:7091/#/Overview
default username and password is admin/admin
3.代碼工程
實(shí)驗(yàn)?zāi)繕?biāo)
訂單服務(wù)調(diào)用庫(kù)存服務(wù)和賬戶余額服務(wù)進(jìn)行相應(yīng)的扣減,并且最終生成訂單
seata-order
訂單服務(wù)
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; // 購(gòu)買數(shù)量,暫時(shí)設(shè)置為 1。
log.info("[createOrder] 當(dāng)前 XID: {}", RootContext.getXID());
// <2> 扣減庫(kù)存
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 {
// 參數(shù)拼接
JSONObject params = new JSONObject().fluentPut("productId", String.valueOf(productId))
.fluentPut("amount", String.valueOf(amount));
// 執(zhí)行調(diào)用
HttpResponse response = DefaultHttpExecutor.getInstance().executePost("http://127.0.0.1:8082", "/stock",
params, HttpResponse.class);
// 解析結(jié)果
Boolean success = Boolean.valueOf(EntityUtils.toString(response.getEntity()));
if (!success) {
throw new RuntimeException("扣除庫(kù)存失敗");
}
}
private void reduceBalance(Long userId, Integer price) throws IOException {
// 參數(shù)拼接
JSONObject params = new JSONObject().fluentPut("userId", String.valueOf(userId))
.fluentPut("price", String.valueOf(price));
// 執(zhí)行調(diào)用
HttpResponse response = DefaultHttpExecutor.getInstance().executePost("http://127.0.0.1:8083", "/balance",
params, HttpResponse.class);
// 解析結(jié)果
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 配置項(xiàng),對(duì)應(yīng) SeataProperties 類
seata:
application-id: ${spring.application.name} # Seata 應(yīng)用編號(hào),默認(rèn)為 ${spring.application.name}
tx-service-group: ${spring.application.name}-group # Seata 事務(wù)組編號(hào),用于 TC 集群名
# 服務(wù)配置項(xiàng),對(duì)應(yīng) ServiceProperties 類
service:
# 虛擬組和分組的映射
vgroup-mapping:
order-service-group: default
# 分組和 Seata 服務(wù)的映射
grouplist:
default: 127.0.0.1:8091seata-product
商品庫(kù)存服務(wù)
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] 收到減少庫(kù)存請(qǐng)求, 商品:{}, 價(jià)格:{}", productReduceStockDTO.getProductId(),
productReduceStockDTO.getAmount());
try {
productService.reduceStock(productReduceStockDTO.getProductId(), productReduceStockDTO.getAmount());
// 正??鄢龓?kù)存,返回 true
return true;
} catch (Exception e) {
// 失敗扣除庫(kù)存,返回 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> 開(kāi)啟新事物
public void reduceStock(Long productId, Integer amount) throws Exception {
log.info("[reduceStock] 當(dāng)前 XID: {}", RootContext.getXID());
// <2> 檢查庫(kù)存
checkStock(productId, amount);
log.info("[reduceStock] 開(kāi)始扣減 {} 庫(kù)存", productId);
// <3> 扣減庫(kù)存
int updateCount = productDao.reduceStock(productId, amount);
// 扣除成功
if (updateCount == 0) {
log.warn("[reduceStock] 扣除 {} 庫(kù)存失敗", productId);
throw new Exception("庫(kù)存不足");
}
// 扣除失敗
log.info("[reduceStock] 扣除 {} 庫(kù)存成功", productId);
}
private void checkStock(Long productId, Integer requiredAmount) throws Exception {
log.info("[checkStock] 檢查 {} 庫(kù)存", productId);
Integer stock = productDao.getStock(productId);
if (stock < requiredAmount) {
log.warn("[checkStock] {} 庫(kù)存不足,當(dāng)前庫(kù)存: {}", productId, stock);
throw new Exception("庫(kù)存不足");
}
}
}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 {
/**
* 獲取庫(kù)存
*
* @param productId 商品編號(hào)
* @return 庫(kù)存
*/
@Select("SELECT stock FROM product WHERE id = #{productId}")
Integer getStock(@Param("productId") Long productId);
/**
* 扣減庫(kù)存
*
* @param productId 商品編號(hào)
* @param amount 扣減數(shù)量
* @return 影響記錄行數(shù)
*/
@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
用戶余額服務(wù)
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] 收到減少余額請(qǐng)求, 用戶:{}, 金額:{}", 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> 開(kāi)啟新事物
public void reduceBalance(Long userId, Integer price) throws Exception {
log.info("[reduceBalance] 當(dāng)前 XID: {}", RootContext.getXID());
// <2> 檢查余額
checkBalance(userId, price);
log.info("[reduceBalance] 開(kāi)始扣減用戶 {} 余額", 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] 用戶 {} 余額不足,當(dāng)前余額:{}", 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 需要扣減的數(shù)目
* @return 影響記錄行數(shù)
*/
@Update("UPDATE account SET balance = balance - #{price} WHERE id = 1 AND balance >= ${price}")
int reduceBalance(@Param("price") Integer price);
}以上只是一些關(guān)鍵代碼,所有代碼請(qǐng)參見(jiàn)下面代碼倉(cāng)庫(kù)
代碼倉(cāng)庫(kù)
https://github.com/Harries/springboot-demo
4.測(cè)試
- 啟動(dòng)seata-order服務(wù)
- 啟動(dòng)seata-product服務(wù)
- 啟動(dòng)seata-balance服務(wù)
?編輯可以看到控制臺(tái)輸出回滾日志
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
5.引用
到此這篇關(guān)于Spring Boot集成Seata實(shí)現(xiàn)基于AT模式的分布式事務(wù)的文章就介紹到這了,更多相關(guān)Spring Boot集成Seata分布式事務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
常用的Spring Boot調(diào)用外部接口方式實(shí)現(xiàn)數(shù)據(jù)交互
Spring Boot提供了多種調(diào)用外部接口的方式,可以方便地實(shí)現(xiàn)與其他系統(tǒng)的數(shù)據(jù)交互,提高系統(tǒng)的可擴(kuò)展性和數(shù)據(jù)共享能力,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)吧2023-04-04
java實(shí)現(xiàn)web實(shí)時(shí)消息推送的七種方案
這篇文章主要為大家介紹了java實(shí)現(xiàn)web實(shí)時(shí)消息推送的七種方案示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2022-07-07
Mybatis-Plus中分頁(yè)插件PaginationInterceptor的使用
我們?cè)陂_(kāi)發(fā)的過(guò)程中,經(jīng)常會(huì)遇到分頁(yè)操作,本文主要介紹了Mybatis-Plus中分頁(yè)插件PaginationInterceptor的使用,文中通過(guò)示例代碼介紹的非常詳細(xì),需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2023-06-06
Java對(duì)稱與非對(duì)稱加密算法原理詳細(xì)講解
對(duì)稱加密算法指加密和解密使用相同密鑰的加密算法。對(duì)稱加密算法用來(lái)對(duì)敏感數(shù)據(jù)等信息進(jìn)行加密,非對(duì)稱加密算法指加密和解密使用不同密鑰的加密算法,也稱為公私鑰加密2022-11-11
Java 注解@NotNull 和 @NotEmpty區(qū)別深度解析
本文給大家介紹了Java注解@NotNull和@NotEmpty區(qū)別深度解析,本文結(jié)合實(shí)例代碼給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友參考下吧2025-09-09
SpringBoot請(qǐng)求轉(zhuǎn)發(fā)的方式小結(jié)
本文主要介紹了SpringBoot請(qǐng)求轉(zhuǎn)發(fā)的方式,一共有兩大類,一種是controller控制器轉(zhuǎn)發(fā)一種是使用HttpServletRequest進(jìn)行轉(zhuǎn)發(fā),本文就詳細(xì)的介紹一下,感興趣的可以了解一下2023-09-09

