Spring事務管理（二）分佈式事務管理之JTA與鏈式事務

時間 2021-08-12

標籤 php java mysql web spring sql 數據庫 springboot app 分佈式欄目 Spring 简体版

原文原文鏈接

什麼是分佈式事務

跨庫的事務就屬於分佈式事務，好比對兩個庫的不一樣表同時修改和同時rollback等。php

上一節中，咱們只是演示了單個庫(數據源)的事務處理。這一節主要講如何處理多個數據源的事務。java

爲何多數據源下不能使用普通事務來處理呢?

我想不少人都有這個問題，打個比方，分庫分表後有個數據庫A和數據庫B，A中有搶票記錄，B中有票數記錄。當咱們完成搶票功能，須要在B減小票數的同時在A中增長記錄。可是若是有下面的代碼發生：mysql

@Transactional

public
 
void
 multiDBTX
(){

    B
.
reduce
(
ticketId
);

    
if
 
(
true
){

        
throw
 
new
 
RuntimeException
(
"throw new exception"
);

    
}

    A
.
save
(
result
);

}

我在B扣除票數後拋出異常，而後執行A庫添加記錄。web

若是沒有分佈式事務處理，則結果就是B票數扣除，但A沒有保存記錄。也就是出錯後B並無進行事務回滾。spring

那問題來了，怎麼才能實現咱們的要求呢。sql

分佈式事務原則

CAP定理

web沒法同時知足如下三點：數據庫

一致性: 全部數據變更都是同步的springboot

可用性: 每一個操做都必須有預期的響應app

分區容錯性: 出現單個節點沒法可用，系統依然正常對外提供服務分佈式

BASE理論

BASE理論是對CAP中的一致性和可用性進行一個權衡的結果。核心思想是即便沒法作到強一致性，但可使用一些技術手段達到最終一致。

Basically Available（基本可用）:容許系統發生故障時，損失一部分可用性。

Soft state（軟狀態）：容許數據同步存在延遲。

Eventually consistent（最終一致性）：不須要保持強一致性，最終一致便可。

那如何來實現分佈式事務管理呢？

分佈式事務管理實踐

1. JTA實現

事務有效的屏蔽了底層事務資源，使應用能夠以透明的方式參入到事務處理中，可是與本地事務相比，XA 協議的系統開銷大

在這裏我先帶你們走出一個誤解，你在網上搜JTA通常都是分佈式事務用它，可是它就是用來作分佈式事務的嗎？不是的，我在上文說過，JTA只是Java實現XA事務的一個規範，咱們在第一節 Spring事務管理（一）快速入門中用到的事務，均可以叫JTA事務管理。下面主要說JTA實現分佈式事務管理：

這裏咱們會用到Atomikos事務管理器，它是一個開源的事務管理器，實現了XA的一種分佈式事務處理並能夠嵌入到你的SpringBoot當中。

拓展：什麼是XA

基本上全部的數據庫都會支持XA事務，百度百科上說法：XA協議由Tuxedo首先提出的，並交給X/Open組織，做爲資源管理器（數據庫）與事務管理器的接口標準。簡單的說，它是事務的標準，JTA也是它標準的java實現。

1.1 導入pom

<dependency>

    
<groupId>
org.springframework.boot
</groupId>

    
<artifactId>
spring-boot-starter-jta-atomikos
</artifactId>

</dependency>

1.2 設置數據源

SpringBoot設置多數據源這裏只說下思路(重點仍是說事務實現)：

在 application.yml中配置多數據源配置。
寫配置類加載配置並放入DataSource並設置事務：

@Configuration

@DependsOn
(
"transactionManager"
)

@EnableJpaRepositories
(
basePackages 
=
 
"com.fantj.repository.user"
,
 entityManagerFactoryRef 
=
 
"userEntityManager"
,
 transactionManagerRef 
=
 
"transactionManager"
)

@EnableConfigurationProperties
(
UserDatasourceProperties
.
class
)

public
 
class
 
UserConfig
 
{




    
@Autowired

    
private
 
JpaVendorAdapter
 jpaVendorAdapter
;




    
// 這裏注入 dataSource信息的類 

    
@Autowired

    
private
 
UserDatasourceProperties
 userDatasourceProperties
;




    
@Bean
(
name 
=
 
"userDataSource"
)

    
public
 
DataSource
 userDataSource
()
 
{

        
// 給XADataSource 設置 DataSource 屬性

        
MysqlXADataSource
 mysqlXaDataSource 
=
 
new
 
MysqlXADataSource
();

        mysqlXaDataSource
.
setURL
(
userDatasourceProperties
.
getUrl
());

        mysqlXaDataSource
.
setUser
(
userDatasourceProperties
.
getUser
());

        mysqlXaDataSource
.
setPassword
(
userDatasourceProperties
.
getPassword
());

        mysqlXaDataSource
.
setPinGlobalTxToPhysicalConnection
(
true
);

        
// 建立 Atomiko， 並將 mysql的XA交給JTA管理

        
AtomikosDataSourceBean
 xaDataSource 
=
 
new
 
AtomikosDataSourceBean
();

        xaDataSource
.
setXaDataSource
(
mysqlXaDataSource
);

        
// 設置惟一資源名

        xaDataSource
.
setUniqueResourceName
(
"datasource2"
);

        
return
 xaDataSource
;

    
}




    
@Bean
(
name 
=
 
"userEntityManager"
)

    
@DependsOn
(
"transactionManager"
)

    
public
 
LocalContainerEntityManagerFactoryBean
 userEntityManager
()
 
throws
 
Throwable
 
{




        
HashMap
<
String
,
 
Object
>
 properties 
=
 
new
 
HashMap
<
String
,
 
Object
>();

        properties
.
put
(
"hibernate.transaction.jta.platform"
,
 
AtomikosJtaPlatform
.
class
.
getName
());

        properties
.
put
(
"javax.persistence.transactionType"
,
 
"JTA"
);




        
LocalContainerEntityManagerFactoryBean
 entityManager 
=
 
new
 
LocalContainerEntityManagerFactoryBean
();

        
// 給工廠bean設置 資源加載屬性

        entityManager
.
setJtaDataSource
(
userDataSource
());

        entityManager
.
setJpaVendorAdapter
(
jpaVendorAdapter
);

        entityManager
.
setPackagesToScan
(
"com.fantj.pojo.user"
);

        entityManager
.
setPersistenceUnitName
(
"userPersistenceUnit"
);

        entityManager
.
setJpaPropertyMap
(
properties
);

        
return
 entityManager
;

    
}




}

這只是一個數據源的配置，第二個數據源的配置也相似，注意不能同Entity同Repository，映射放在不一樣包下實現。兩個都返回LocalContainerEntityManagerFactoryBean它便會交給@Transaction去管理，兩個數據源配置完後。這樣的代碼B將會回滾。

@Transactional

public
 
void
 multiDBTX
(){

    B
.
reduce
(
ticketId
);

    
if
 
(
true
){

        
throw
 
new
 
RuntimeException
(
"throw new exception"
);

    
}

    A
.
save
(
result
);

}

1.3 JTA缺點

由於JTA採用兩階段提交方式，第一次是預備階段，第二次纔是正式提交。當第一次提交出現錯誤，則整個事務出現回滾，一個事務的時間可能會較長，由於它要跨越多個數據庫多個數據資源的的操做，因此在性能上可能會形成吞吐量低。並且，它只能用在單個服務內。一個完善的JTA事務還須要同時考慮不少元素，這只是個示例。

2. 鏈式事務管理

鏈式事務就是聲明一個ChainedTransactionManager 將全部的數據源事務按順序放到該對象中，則事務會按相反的順序來執行事務。

網上發現了一個鏈式事務管理的處理順序，總結的很到位。

1.start
 message transaction

2.receive
 message

3.start
 database transaction

4.update
 database

5.commit
 database transaction

6.commit
 message transaction   
##當這一步出現錯誤時，上面的由於已經commit，因此不會rollback

能夠看到，從345能夠看到，它後拿到的事務先提交，這就致使若是1出錯，則不會進行數據回滾。跟Spring的同步事務差很少，同步事務也是這種特性。

下面我會測試這個性質。

爲了方便，我拿JdbcTemplate來測試該事務。

DBConfig.java

配置DataSource以及返回Template新實例和鏈式事務配置。

/**

 * DB配置類

 */

@Configuration

public
 
class
 
DBConfig
 
{

    
/**

     * user-DB配置

     */

    
@Bean

    
@Primary

    
@ConfigurationProperties
(
prefix 
=
 
"spring.datasource.user"
)

    
public
 
DataSourceProperties
 userDataSourceProperties
(){

        
return
 
new
 
DataSourceProperties
();

    
}




    
@Bean

    
@Primary

    
public
 
DataSource
 userDataSource
(){

        
return
 userDataSourceProperties
().
initializeDataSourceBuilder
().
type
(
HikariDataSource
.
class
).
build
();

    
}




    
@Bean

    
public
 
JdbcTemplate
 userJdbcTemplate
(
@Qualifier
(
"userDataSource"
)
 
DataSource
 userDataSource
){

        
return
 
new
 
JdbcTemplate
(
userDataSource
);

    
}

    
/**

     * result-DB配置

     */

    
@Bean

    
@ConfigurationProperties
(
prefix 
=
 
"spring.datasource.result"
)

    
public
 
DataSourceProperties
 resultDataSourceProperties
(){

        
return
 
new
 
DataSourceProperties
();

    
}




    
@Bean

    
public
 
DataSource
 resultDataSource
(){

        
return
 resultDataSourceProperties
().
initializeDataSourceBuilder
().
type
(
HikariDataSource
.
class
).
build
();

    
}




    
@Bean

    
public
 
JdbcTemplate
 resultJdbcTemplate
(
@Qualifier
(
"resultDataSource"
)
 
DataSource
 resultDataSource
){

        
return
 
new
 
JdbcTemplate
(
resultDataSource
);

    
}

    
/**

     * 鏈式事務配置

     */

    
@Bean

    
public
 
PlatformTransactionManager
 transactionManager
(){

        
DataSourceTransactionManager
 userTM 
=
 
new
 
DataSourceTransactionManager
(
userDataSource
());

        
DataSourceTransactionManager
 resultTM 
=
 
new
 
DataSourceTransactionManager
(
resultDataSource
());

        
return
 
new
 
ChainedTransactionManager
(
userTM
,
resultTM
);

    
}

}

transactionManager()方法實現了鏈式事務配置，注意我放置的順序先userTM後resultTM，因此事務應該是先拿到 userTM而後拿到 resultTM而後提交 resultTM最後提交 userTM，也就是說，若是我在提交user事務的時候出錯，此時result相關的事務已經提交完成，因此result數據是不能回滾的。

2.1 測試

@RequestMapping
(
""
)

@Transactional

public
 
void
 testTX
(){

//        resultJdbcTemplate.execute("insert into result values(68,6,6)");

    userJdbcTemplate
.
execute
(
"insert into user values (6,'FantJ',23,'男')"
);

    
if
 
(
true
){

        
throw
 
new
 
RuntimeException
(
"yes , throw one exception"
);

    
}

    resultJdbcTemplate
.
execute
(
"insert into result values(66,6,6)"
);

//        userJdbcTemplate.execute("insert into user values (8,'FantJ',23,'男')");

}

兩個數據庫沒有內容。

控制檯精簡後的日誌：

Creating
 
new
 transaction 
with
 name 
[
springbootjtamultidb
.
jtamulti
.
JtaMultiApplicationTests
.
testTX
]:
 

Creating
 
new
 transaction 
with
 name 
[
springbootjtamultidb
.
jtamulti
.
JtaMultiApplicationTests
.
testTX
]:
 

Began
 transaction 
(
1
)
 
for
 test context 
[
DefaultTestContext@1dde4cb2
 testClass 
=
 
...

Executing
 SQL statement 
[
insert 
into
 user values 
(
6
,
'FantJ'
,
23
,
'男'
)]

Initiating
 transaction rollback

Rolling
 back JDBC transaction on 
Connection
 
[
HikariProxyConnection@318550723
 wrapping com
.
mysql
.
cj
.
jdbc
.
ConnectionImpl@57bd6a8f
]

Releasing
 JDBC 
Connection
 
[
HikariProxyConnection@318550723
 wrapping com
.
mysql
.
cj
.
jdbc
.

Initiating
 transaction rollback

Rolling
 back JDBC transaction on 
Connection
 
[
HikariProxyConnection@1201991394
 wrapping com
.
mysql
.
cj
.
jdbc
.
ConnectionImpl@36f6e521
]

Releasing
 JDBC 
Connection
 
[
HikariProxyConnection@1201991394
 wrapping com
.
mysql
.
cj
.
jdbc
.
ConnectionImpl@36f6e521
]
 after transaction

Resuming
 suspended transaction after completion of inner transaction

Rolled
 back transaction 
for
 test
:
 
[
DefaultTestContext@1dde4cb2
 testClass 
=
 
JtaMultiApplicationTests
,
 testInstance 
=
 springbootjtamultidb
.
jtamulti
.
JtaMultiApplicationTests@441772e

其實只要是兩個dao操做中間出錯或者第一個dao操做以前出錯，事務都能正常回滾。若是result操做再前，user操做再後，user操做完拋出異常，也能回滾事務，緣由上文有講。

@RequestMapping
(
""
)

@Transactional

public
 
void
 testTX
(){

    resultJdbcTemplate
.
execute
(
"insert into result values(68,6,6)"
);

//        userJdbcTemplate.execute("insert into user values (6,'FantJ',23,'男')");

//        if (true){

//            throw new RuntimeException("yes , throw one exception");

//        }

//        resultJdbcTemplate.execute("insert into result values(66,6,6)");

    userJdbcTemplate
.
execute
(
"insert into user values (8,'FantJ',23,'男')"
);

    
if
 
(
true
){

        
throw
 
new
 
RuntimeException
(
"yes , throw one exception"
);

    
}

}

這段代碼也能正常回滾，結果我就不貼了。(浪費你們精力)

2.2 驗證第二個事務不能回滾的狀況

重要的事情再重複一遍:注意我放置的順序先userTM後resultTM，因此事務應該是先拿到 userTM而後拿到 resultTM而後提交 resultTM最後提交 userTM，也就是說，若是我在提交user事務的時候出錯，此時result相關的事務已經提交完成，因此result數據是不能回滾的。

代碼和以前的同樣，須要在事務提交的方法中打斷點

@RequestMapping
(
""
)

@Transactional

public
 
void
 testTX
(){

    resultJdbcTemplate
.
execute
(
"insert into result values(68,6,6)"
);

    userJdbcTemplate
.
execute
(
"insert into user values (8,'FantJ',23,'男')"
);

}

注意攔截到斷點時，先放行一個commit，也就是result事務的commit，而後攔截到第二個commit請求時，關閉user所在的數據庫，而後放行。

下面是將第一個commit請求放行後的控制檯日誌：

Creating
 
new
 transaction 
with
 name 
[
springbootjtamultidb
.
jtamulti
.
controller
.
MainController
.

Acquired
 
Connection
 
[
HikariProxyConnection@810768078
 wrapping com
.
mysql
.
cj
.
jdbc
.
ConnectionImpl@3cfb22cd
]
 
for
 JDBC transaction

Switching
 JDBC 
Connection
 
[
HikariProxyConnection@810768078
 wrapping com
.
mysql
.
cj
.
jdbc
.
ConnectionImpl@3cfb22cd
]
 to manual commit

Executing
 SQL statement 
[
insert 
into
 result values
(
68
,
6
,
6
)]

Executing
 SQL statement 
[
insert 
into
 user values 
(
8
,
'FantJ'
,
23
,
'男'
)]

Initiating
 transaction commit

Committing
 JDBC transaction on 
Connection
 
[
HikariProxyConnection@810768078
 wrapping com
.
mysql
.
cj
.
jdbc
.
ConnectionImpl@3cfb22cd
]

 
Releasing
 JDBC 
Connection
 
[
HikariProxyConnection@810768078
 wrapping com
.
mysql
.
cj
.
jdbc
.
ConnectionImpl@3cfb22cd
]
 after transaction

 
Resuming
 suspended transaction after completion of inner transactio

n

注意倒數第三行日誌，它出現證實咱們第一個result的sql事務已經提交，此時你刷新數據庫數據已經更新了，可是咱們斷點並尚未放行，user的事務尚未提交，我把user的數據庫源關閉，再放行，能夠看到，result已經有數據，user沒有數據，此時result並無進行回滾，這是鏈式事務的缺點。