分布式事務解决方案:XA規範

Java識堂 2021-09-18 17:20:28 阅读数:372

分布式 分布 解决方案 解决 方案

請添加圖片描述

XA規範

二階段提交協議是一個協議,而XA規範是X/Open 組織針對二階段提交協議的實現做的規範。目前幾乎所有的主流數據庫都對XA規範提供了支持。

這樣做的好處是方便多個資源(如數據庫,應用服務器,消息隊列等)在同一個事務中訪問。你可以類比JDBC

我們這篇文章就以MySQL XA為例演示一下XA怎麼玩?

MySQL XA常用的命令如下

命令 解釋
XA START xid 開啟一個事務,並將事務置於ACTIVE狀態,此後執行的SQL語句都將置於該事務中
XA END xid 將事務置於IDLE狀態,錶示事務內的SQL操作完成
XA PREPARE xid 實現事務提交的准備工作,事務狀態置於PREPARED狀態。事務如果無法完成提交前的准備操作,該語句會執行失敗
XA COMMIT xid 事務最終提交,完成持久化
XA ROLLBACK xid 事務回滾終止
XA RECOVER 查看MySQL中存在的PREPARED狀態的xa事務

我們在db_account_1和db_account_2都建一個account_info錶並初始化2條記錄

CREATE TABLE `account_info`
(
`id` INT(11) NOT NULL AUTO_INCREMENT COMMENT '自增主鍵',
`user_id` VARCHAR(255) NOT NULL COMMENT '用戶id',
`balance` INT(11) NOT NULL DEFAULT 0 COMMENT '用戶餘額',
PRIMARY KEY (`id`)
) ENGINE = InnoDB
AUTO_INCREMENT = 1
DEFAULT CHARSET = utf8;
INSERT INTO account_info (id, user_id, balance)
VALUES (1, '1001', 10000);
INSERT INTO account_info (id, user_id, balance)
VALUES (2, '1002', 10000);

我們以用戶1001向1002轉賬200元為例

mysql> XA START "transfer_money";
Query OK, 0 rows affected (0.01 sec)
mysql> update account_info set balance = balance - 200 where user_id = '1001';
Query OK, 1 row affected (0.01 sec)
Rows matched: 1 Changed: 1 Warnings: 0
mysql> XA END "transfer_money";
Query OK, 0 rows affected (0.01 sec)
mysql> XA PREPARE "transfer_money";
Query OK, 0 rows affected (0.01 sec)
mysql> XA COMMIT "transfer_money";
Query OK, 0 rows affected (0.01 sec)

在XA START執行後所有資源將會被鎖定,直到執行了XA PREPARE或者XA COMMIT才會釋放。

如果在這個時間段內另外一個事務執行如下語句則會一直被阻塞

update account_info set balance = balance - 200 where user_id = '1001';

這就是XA規範這種解决方案很少被使用的原因,因為中間過程會鎖定資源,很難支持高並發

我們也可以將一個 IDLE 狀態的 XA 事務可以直接提交或者回滾

mysql> XA COMMIT "transfer_money";
1399 - XAER_RMFAIL: The command cannot be executed when global transaction is in the IDLE state
mysql> XA COMMIT "transfer_money" ONE PHASE;
Query OK, 0 rows affected (0.01 sec)
mysql> XA START "transfer_money";
Query OK, 0 rows affected (0.01 sec)
mysql> update account_info set balance = balance - 200 where user_id = '1001';
Query OK, 1 row affected (0.01 sec)
Rows matched: 1 Changed: 1 Warnings: 0
mysql> XA END "transfer_money";
Query OK, 0 rows affected (0.01 sec)
mysql> XA COMMIT "transfer_money" ONE PHASE;
Query OK, 0 rows affected (0.01 sec)

XA事務變化圖
在這裏插入圖片描述

JTA

JTA(Java Transaction API),是J2EE的編程接口規範,它是XA規範的Java實現相關的接口有如下2個

javax.transaction.TransactionManager(事務管理器的接口):定義了有關事務的開始、提交、撤回等操作。
javax.transaction.xa.XAResource(滿足XA規範的資源定義接口):一種資源如果要支持JTA事務,就需要讓它的資源實現該XAResource接口,並實現該接口定義的兩階段提交相關的接口

在Java中有很多框架都對XA規範進行了實現,我就演示一下最常用的實現atomikos和seata

atomikos只能用在單個應用對多個庫進行操作的場景。而seata所有的分布式事務場景都能用
是什麼造成這種差异呢?看Demo

atomikos

先加依賴

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jta-atomikos</artifactId>
<version>2.1.14.RELEASE</version>
</dependency>

配置2個數據源

spring:
jta:
atomikos:
datasource:
primary:
borrow-connection-timeout: 10000.0
max-lifetime: 20000.0
max-pool-size: 25.0
min-pool-size: 3.0
unique-resource-name: test1
xa-data-source-class-name: com.mysql.jdbc.jdbc2.optional.MysqlXADataSource
xa-properties:
password: test
url: jdbc:mysql://myhost:3306/db_account_1
user: test
secondary:
borrow-connection-timeout: 10000.0
max-lifetime: 20000.0
max-pool-size: 25.0
min-pool-size: 3.0
unique-resource-name: test2
xa-data-source-class-name: com.mysql.jdbc.jdbc2.optional.MysqlXADataSource
xa-properties:
password: test
url: jdbc:mysql://myhost:3306/db_account_2
user: test
enabled: true
@Configuration
public class DataSourceConfig {

@Bean
@ConfigurationProperties(prefix = "spring.jta.atomikos.datasource.primary")
public DataSource primaryDataSource() {

return new AtomikosDataSourceBean();
}
@Bean
@ConfigurationProperties(prefix = "spring.jta.atomikos.datasource.secondary")
public DataSource secondaryDataSource() {

AtomikosDataSourceBean ds = new AtomikosDataSourceBean();
return ds;
}
@Bean
public JdbcTemplate primaryJdbcTemplate(
@Qualifier("primaryDataSource") DataSource dataSource) {

return new JdbcTemplate(dataSource);
}
@Bean
public JdbcTemplate secondaryJdbcTemplate(
@Qualifier("secondaryDataSource") DataSource dataSource) {

return new JdbcTemplate(dataSource);
}
}
@Service
public class AccountService {

@Resource
@Qualifier("primaryJdbcTemplate")
private JdbcTemplate primaryJdbcTemplate;
@Resource
@Qualifier("secondaryJdbcTemplate")
private JdbcTemplate secondaryJdbcTemplate;
@Transactional(rollbackFor = Exception.class)
public void tx1() {

Integer money = 100;
String sql = "update account_info set balance = balance + ? where user_id = ?";
primaryJdbcTemplate.update(sql, new Object[]{
-money, 1001});
secondaryJdbcTemplate.update(sql, new Object[]{
money, 1002});
}
@Transactional(rollbackFor = Exception.class)
public void tx2() {

Integer money = 100;
String sql = "update account_info set balance = balance + ? where user_id = ?";
primaryJdbcTemplate.update(sql, new Object[]{
-money, 1001});
secondaryJdbcTemplate.update(sql, new Object[]{
money, 1002});
throw new RuntimeException();
}
}
@RunWith(SpringRunner.class)
@SpringBootTest
public class AtomikosAtApplicationTests {

@Resource
private AccountService accountService;
// 正常執行
@Test
public void test1() {

accountService.tx1();
}
// 异常回滾
@Test
public void test2() {

accountService.tx2();
}
}

seata

我們需要開發2個應用,就不貼pom依賴了,從github看依賴吧

https://github.com/erlieStar/spring-cloud-distributed-transaction

seata-xa-tm

我們只需要配置一下application.yaml即可。你可能看到很多文章還需要配置file.conf和registry.conf,用了spring starter後直接在application.yaml配置即可

application.yaml

server:
port: 30002
spring:
application:
name: seata-xa-tm
datasource:
driver-class-name: com.mysql.jdbc.Driver
url : jdbc:mysql://myhost:3306/db_account_1?useUnicode=true&characterEncoding=utf8
username: test
password: test
seata:
data-source-proxy-mode: XA
enabled: true
application-id: ${
spring.application.name}
tx-service-group: my_test_tx_group
service:
vgroup-mapping:
my_test_tx_group: default
grouplist:
default: myhost:18091
disable-global-transaction: false
config:
type: file
file:
name: file.conf
registry:
type: file
file:
name: file.conf

用@EnableAutoDataSourceProxy注解開啟數據源代理,只需指定為XA模式,因為默認是AT模式

@EnableFeignClients
@SpringBootApplication
@EnableAutoDataSourceProxy(dataSourceProxyMode = "XA")
public class SeataXATm {

public static void main(String[] args) {

SpringApplication.run(SeataXATm.class, args);
}
}

開發轉賬接口

@RestController
@RequestMapping("account")
public class AccountController {

@Resource
private JdbcTemplate jdbcTemplate;
@Resource
private RmAccountClient rmAccountClient;
@GlobalTransactional
@RequestMapping("transfer")
public String transfer(@RequestParam("fromUserId") String fromUserId,
@RequestParam("toUserId") String toUserId,
@RequestParam("money") Integer money,
@RequestParam(value = "flag", required = false) Boolean flag) {

String sql = "update account_info set balance = balance + ? where user_id = ?";
jdbcTemplate.update(sql, new Object[]{
-money, fromUserId});
String result = rmAccountClient.transfer(fromUserId, toUserId, money);
if ("fail".equals(result)) {

throw new RuntimeException("轉賬失敗");
}
if (flag != null && flag) {

throw new RuntimeException("測試同時回滾");
}
return "success";
}
}

調用另外一個賬戶服務,為了方便我就不用注册中心了,直接指定了服務的地址

@FeignClient(value = "seata-xa-rm", url = "http://127.0.0.1:30001")
public interface RmAccountClient {

@RequestMapping("account/transfer")
String transfer(@RequestParam("fromUserId") String fromUserId,
@RequestParam("toUserId") String toUserId,
@RequestParam("money") Integer money);
}

seata-xa-rm

這是我們開發的另一個賬戶服務

application.yaml

server:
port: 30001
spring:
application:
name: seata-xa-rm
datasource:
driver-class-name: com.mysql.jdbc.Driver
url : jdbc:mysql://myhost:3306/db_account_2?useUnicode=true&characterEncoding=utf8
username: test
password: test
seata:
data-source-proxy-mode: XA
enabled: true
application-id: ${
spring.application.name}
tx-service-group: my_test_tx_group
service:
vgroup-mapping:
my_test_tx_group: default
grouplist:
default: myhost:18091
disable-global-transaction: false
config:
type: file
file:
name: file.conf
registry:
type: file
file:
name: file.conf

啟動類

@SpringBootApplication
@EnableAutoDataSourceProxy(dataSourceProxyMode = "XA")
public class SeataXARm {

public static void main(String[] args) {

SpringApplication.run(SeataXARm.class, args);
}
}

轉賬接口

@RestController
@RequestMapping("account")
public class AccountController {

@Resource
private JdbcTemplate jdbcTemplate;
@RequestMapping("transfer")
public String transfer(@RequestParam("fromUserId") String fromUserId,
@RequestParam("toUserId") String toUserId,
@RequestParam("money") Integer money) {

String sql = "update account_info set balance = balance + ? where user_id = ?";
int result = jdbcTemplate.update(sql, new Object[]{
money, toUserId});
if (result == 0) {

return "fail";
}
return "success";
}
}

測試

啟動這2個服務

測試正常轉賬

curl http:127.0.0.1:30002/account?fromUserId=1001&toUserId=1002&money=100

測試seata-xa-tm項目失敗回滾

curl http:127.0.0.1:30002/account?fromUserId=1001&toUserId=1002&money=100&flag=truee

用flag=true來讓seata-xa-tm項目失敗回滾

測試seata-xa-rm項目失敗回滾

curl http:127.0.0.1:30002/account?fromUserId=1001&toUserId=1003&money=100&flag=truee

toUserId=1003,用戶不存在,seata-xa-rm返回fail回滾

參考博客

匯總
[1]https://segmentfault.com/a/1190000040321750
[2]https://zhuanlan.zhihu.com/p/183753774
xa事務
[3]https://www.jianshu.com/p/a59c79186b6d
[4]https://www.jianshu.com/p/7003d58ea182
jta
[5]https://www.jianshu.com/p/86b4ab4f2d18

版权声明:本文为[Java識堂]所创,转载请带上原文链接,感谢。 https://gsmany.com/2021/09/20210918172027752t.html