知識庫 / Spring RSS 訂閱

1. 引言

在本教程中，我們將瞭解在 Java 中“事務”的含義。通過此，我們將理解如何執行本地事務和全局事務。這將使我們能夠探索在 Java 和 Spring 中管理事務的不同方法。

2. 什麼是事務？

在Java中，事務是指一系列操作，必須全部成功完成。因此，如果其中一個或多個操作失敗，則所有其他操作必須回滾，從而使應用程序的狀態保持不變。 這樣做是為了確保應用程序狀態的完整性不會受到損害。

此外，這些事務可能涉及一個或多個資源，例如數據庫、消息隊列，從而產生不同的事務執行方式。 這些方式包括執行本地事務，即針對單個資源執行事務。 也可以讓多個資源參與全局事務。

3. 資源本地事務

我們將首先探討如何在 Java 中使用事務，同時處理單個資源。在這裏，我們可能需要執行多個與單個資源（例如數據庫）相關的操作。但是，我們可能希望這些操作作為一個整體進行，就像一個不可分割的工作單元。換句話説，我們希望這些操作在單個事務下執行。

在 Java 中，我們有多種訪問和操作資源的途徑，例如數據庫。因此，我們處理事務的方式也並非相同。在本節中，我們將瞭解如何使用一些在 Java 中常用的庫來執行事務。

3.1. JDBC

Java Database Connectivity (JDBC) 是 Java 中定義如何訪問數據庫的 API。不同的數據庫供應商為以供應商無關的方式連接數據庫提供 JDBC 驅動程序。因此，我們從驅動程序中檢索一個 Connection 以執行數據庫上的不同操作：

JDBC 提供了在事務中執行語句的選項。 Connection 的默認行為是自動提交。 換句話説，每個語句都視為一個事務，並在執行後自動提交。

但是，如果我們希望將多個語句組合在一個事務中，這也是可以實現的：

Connection connection = DriverManager.getConnection(CONNECTION_URL, USER, PASSWORD);
try {
    connection.setAutoCommit(false);
    PreparedStatement firstStatement = connection .prepareStatement("firstQuery");
    firstStatement.executeUpdate();
    PreparedStatement secondStatement = connection .prepareStatement("secondQuery");
    secondStatement.executeUpdate();
    connection.commit();
} catch (Exception e) {
    connection.rollback();
}

在這裏，我們已禁用 Connection 的自動提交模式。因此，我們可以手動定義事務邊界並執行 提交 或 回滾 操作。 JDBC 也允許我們設置 保存點，這使我們能夠更好地控制回滾的範圍。

3.2. JPA

Java 持久化 API (JPA) 是一種 Java 規範，可用於彌合面向對象領域模型與關係數據庫系統之間的差距。因此，諸如 Hibernate、EclipseLink 和 iBatis 等第三方提供的 JPA 的多個實現可用。

在 JPA 中，我們可以將常規類定義為實體，從而為它們提供持久身份。 EntityManager 類提供了一個用於在持久性上下文中處理多個實體的必要接口。持久性上下文可以被認為是第一級緩存，其中實體進行管理：

這裏的持久性上下文可以是事務範圍或擴展範圍。 事務範圍的持久性上下文與單個事務綁定。 而擴展範圍的持久性上下文可以跨多個事務擴展。 持久性上下文的默認範圍是事務範圍。

讓我們看看如何手動創建 EntityManager 並定義事務邊界：

EntityManagerFactory entityManagerFactory = Persistence.createEntityManagerFactory("jpa-example");
EntityManager entityManager = entityManagerFactory.createEntityManager();
try {
    entityManager.getTransaction().begin();
    entityManager.persist(firstEntity);
    entityManager.persist(secondEntity);
    entityManager.getTransaction().commit();
} catch (Exception e) {
    entityManager.getTransaction().rollback();
}

在這裏，我們正在創建一個 EntityManager，該 EntityManager 是從 EntityManagerFactory 創建的，並且在事務範圍的持久化上下文中進行操作。然後，我們使用 begin、commit 和 rollback 方法定義事務邊界。

3.3. JMS

Java 消息服務 (JMS) 是一種 Java 規範，它 允許應用程序使用消息進行異步通信。 該 API 允許我們創建、發送、接收和讀取消息，這些消息來自隊列或主題。 遵循 JMS 規範的多個消息服務包括 OpenMQ 和 ActiveMQ。

JMS API 支持將多個發送或接收操作組合在一個事務中。 但是，由於消息集成架構的本質，消息的生產和消費不能屬於同一個事務。 事務的範圍仍然在客户端和 JMS 提供商之間。

JMS 允許我們從供應商特定的 ConnectionFactory 中獲取的 Connection 創建一個 Session。 我們有選擇創建一個事務性的或非事務性的 Session 的選項。 對於非事務性的 Session，我們還可以定義適當的確認模式。

讓我們看看如何創建一個事務性的 Session 以在事務下發送多個消息：

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(CONNECTION_URL);
Connection connection = = connectionFactory.createConnection();
connection.start();
try {
    Session session = connection.createSession(true, 0);
    Destination = destination = session.createTopic("TEST.FOO");
    MessageProducer producer = session.createProducer(destination);
    producer.send(firstMessage);
    producer.send(secondMessage);
    session.commit();
} catch (Exception e) {
    session.rollback();
}

在這裏，我們正在創建一個 MessageProducer，用於指定 Destination 類型的主題。我們從先前創建的 Session 中獲取 Destination，並使用 Session 通過 commit 和 rollback 方法定義事務邊界。

4. 全局交易

正如我們所見，資源本地事務允許我們在單個資源中執行多個操作，作為一個整體。但是，我們經常處理 跨多個資源的運算。例如，在兩個不同的數據庫或數據庫和消息隊列中的操作。在這種情況下，資源本地事務對於我們來説是不夠的。

我們需要在這些場景中 一個全局機制來界定跨多個參與資源的事務。這通常被稱為分佈式事務，並且已經有規範被提出來有效地處理它們。

XA 規範是其中一種規範，它定義了一個事務管理器來控制跨多個資源的事務。Java 通過 JTA 和 JTS 組件對符合 XA 規範的分佈式事務提供了相當成熟的支持。

4.1. JTA

Java Transaction API (JTA) 是在 Java Community Process 框架下開發的 Java Enterprise Edition API。它使 Java 應用程序和應用服務器能夠跨 XA 資源執行分佈式事務。JTA 遵循 XA 架構，採用兩階段提交機制。

JTA 規範了事務管理器與分佈式事務中其他參與者之間的標準 Java 接口：

讓我們來理解一下上面所強調的關鍵接口：

• TransactionManager: 該接口允許應用服務器界定和控制事務
• UserTransaction: 該接口允許應用程序程序明確地界定和控制事務
• XAResource: 該接口的目的是允許事務管理器與 XA 兼容資源的資源管理器協作

4.2. JTS

Java Transaction Service (JTS) 是 與 OMG OTS 規範對應的一種構建事務管理器規範。JTS 使用標準 CORBA ORB/TS 接口和 Internet Inter-ORB Protocol (IIOP) 來在 JTS 事務管理器之間進行事務上下文傳播。

在較高層面上，它支持 Java Transaction API (JTA)。一個 JTS 事務管理器為參與分佈式事務的各方提供事務服務：

JTS 為應用程序提供的服務在很大程度上是透明的，因此我們可能不會在應用程序架構中注意到它們。JTS 以應用程序服務器為基礎，從應用程序程序中抽象出所有事務語義。

5. JTA 事務管理

現在，我們將瞭解如何使用 JTA 管理分佈式事務。分佈式事務並非易事，因此也存在成本影響。此外，我們可以選擇多種方式將 JTA 集成到我們的應用程序中。因此，我們的選擇必須基於整體應用程序架構和目標。

5.1. 在應用服務器中的 JTA

正如我們之前所見，JTA 架構依賴於應用服務器來協調一系列事務相關的操作。它主要依賴服務器提供的是通過 JNDI 的命名服務。這其中，如數據源等 XA 資源會被綁定並從服務器中檢索出來。

除了這一點，我們在應用中管理事務邊界的選擇也至關重要。這導致了在 Java 應用服務器中兩種類型的事務：

• 容器管理事務：正如其名稱所示，在這裏，事務邊界由應用服務器設置。這簡化了企業 JavaBean (EJB) 的開發，因為它不包含與事務界定相關的語句，並完全依賴容器來完成，但這種方式缺乏足夠的靈活性。
• Bean 管理事務：與容器管理事務相反，在 Bean 管理事務中，EJB 包含明確的語句來定義事務界定。這為應用程序提供了精確的控制，可以在標記事務邊界的同時，也帶來了更高的複雜性。

在應用服務器的上下文中執行事務的主要缺點是，應用程序與服務器緊密耦合。這會對應用程序的可測試性、可管理性和可移植性產生影響。在微服務架構中，這種影響更為顯著，因為重點在於開發與服務器無關的應用程序。

5.2. JTA 獨立部署

上一節討論的問題為創建不依賴應用服務器的分佈式事務解決方案提供了強大的動力。 在此方面，我們有多種選擇，例如使用 Spring 的事務支持或使用像 Atomikos 這樣的事務管理器。

讓我們看看如何使用像 Atomikos 這樣的事務管理器來促進數據庫和消息隊列之間的分佈式事務。分佈式事務的關鍵方面是使用事務管理器註冊和註銷參與資源。 Atomikos 已經幫我們處理了這一項工作。 我們只需要使用 Atomikos 提供的抽象即可：

AtomikosDataSourceBean atomikosDataSourceBean = new AtomikosDataSourceBean();
atomikosDataSourceBean.setXaDataSourceClassName("com.mysql.cj.jdbc.MysqlXADataSource");
DataSource dataSource = atomikosDataSourceBean;

在這裏，我們正在創建 AtomikosDataSourceBean 實例並註冊 vendor 級別的 XADataSource。 從這裏開始，我們可以像任何其他 DataSource 一樣使用它，並獲得分佈式事務的好處。

同樣，我們 擁有消息隊列的抽象，它會自動註冊 vendor 級別的 XA 資源與事務管理器。

AtomikosConnectionFactoryBean atomikosConnectionFactoryBean = new AtomikosConnectionFactoryBean();
atomikosConnectionFactoryBean.setXaConnectionFactory(new ActiveMQXAConnectionFactory());
ConnectionFactory connectionFactory = atomikosConnectionFactoryBean;

在這裏，我們正在創建一個 AtomikosConnectionFactoryBean 的實例，並從啓用了 XA 的 JMS 供應商註冊 XAConnectionFactory。之後，我們可以將其作為常規 ConnectionFactory 使用。

現在，Atomikos 提供了將所有內容整合在一起的最後一塊拼圖，即 UserTransaction 的實例：

UserTransaction userTransaction = new UserTransactionImp();

現在，我們準備創建一個跨數據庫和消息隊列的分佈式事務應用程序：

try {
    userTransaction.begin();

    java.sql.Connection dbConnection = dataSource.getConnection();
    PreparedStatement preparedStatement = dbConnection.prepareStatement(SQL_INSERT);
    preparedStatement.executeUpdate();

    javax.jms.Connection mbConnection = connectionFactory.createConnection();
    Session session = mbConnection.createSession(true, 0);
    Destination destination = session.createTopic("TEST.FOO");
    MessageProducer producer = session.createProducer(destination);
    producer.send(MESSAGE);

    userTransaction.commit();
} catch (Exception e) {
    userTransaction.rollback();
}

在這裏，我們使用 UserTransaction 類中的 <strong>begin</strong> 和 <em>commit</em> 方法來界定事務邊界。這包括將記錄保存到數據庫以及將消息發佈到消息隊列。

6. Spring 中的事務支持

我們知道處理事務通常是一項複雜任務，需要大量的樣板代碼和配置。 此外，每個資源都有自己處理本地事務的方式。 在 Java 中，JTA 幫助我們抽象這些差異，但同時也引入了提供商特定的細節以及應用程序服務器的複雜性。

Spring 平台提供了一種更簡潔的方式來處理事務，無論是本地資源事務還是全局事務。 結合 Spring 的其他優勢，使用 Spring 處理事務具有説服力。 此外，使用 Spring 配置和切換事務管理器非常容易，該事務管理器可以是服務器提供的，也可以是獨立的。

Spring 通過為具有事務代碼的方法創建代理來實現這種無縫抽象。 代理代表代碼管理事務狀態，藉助 TransactionManager：

在此界面中，主要接口是 PlatformTransactionManager，它提供了多種實現。 它為 JDBC（DataSource）、JMS、JPA、JTA 以及許多其他資源提供了抽象。

6.1. 配置

讓我們看看如何配置 Spring 使用 Atomikos 作為事務管理器，併為 JPA 和 JMS 提供事務支持。 我們將首先定義一個 JTA 類型的 PlatformTransactionManager：

@Bean
public PlatformTransactionManager platformTransactionManager() throws Throwable {
    return new JtaTransactionManager(
                userTransaction(), transactionManager());
}

在這裏，我們提供了一些 UserTransaction 和 TransactionManager 的實例，供 JTATransactionManager 使用。這些實例是由像 Atomikos 這樣的事務管理器庫提供的：

@Bean
public UserTransaction userTransaction() {
    return new UserTransactionImp();
}

@Bean(initMethod = "init", destroyMethod = "close")
public TransactionManager transactionManager() {
    return new UserTransactionManager();
}

Atomikos 提供了 UserTransactionImp 和 UserTransactionManager 這兩個類。

此外，我們需要定義 JmsTemplete，它是 Spring 中允許同步 JMS 訪問的核心類。

@Bean
public JmsTemplate jmsTemplate() throws Throwable {
    return new JmsTemplate(connectionFactory());
}

在這裏，Atomikos ConnectionFactory 提供了一種解決方案，它支持通過它提供的 Connection 進行分佈式事務。

@Bean(initMethod = "init", destroyMethod = "close")
public ConnectionFactory connectionFactory() {
    ActiveMQXAConnectionFactory activeMQXAConnectionFactory = new 
ActiveMQXAConnectionFactory();
    activeMQXAConnectionFactory.setBrokerURL("tcp://localhost:61616");
    AtomikosConnectionFactoryBean atomikosConnectionFactoryBean = new AtomikosConnectionFactoryBean();
    atomikosConnectionFactoryBean.setUniqueResourceName("xamq");
    atomikosConnectionFactoryBean.setLocalTransactionMode(false);
atomikosConnectionFactoryBean.setXaConnectionFactory(activeMQXAConnectionFactory);
    return atomikosConnectionFactoryBean;
}

@Bean
public LocalContainerEntityManagerFactoryBean entityManager() throws SQLException {
    LocalContainerEntityManagerFactoryBean entityManager = new LocalContainerEntityManagerFactoryBean();
    entityManager.setDataSource(dataSource());
    Properties properties = new Properties();
    properties.setProperty( "javax.persistence.transactionType", "jta");
    entityManager.setJpaProperties(properties);
    return entityManager;
}

@Bean(initMethod = "init", destroyMethod = "close")
public DataSource dataSource() throws SQLException {
    MysqlXADataSource mysqlXaDataSource = new MysqlXADataSource();
    mysqlXaDataSource.setUrl("jdbc:mysql://127.0.0.1:3306/test");
    AtomikosDataSourceBean xaDataSource = new AtomikosDataSourceBean();
    xaDataSource.setXaDataSource(mysqlXaDataSource);
    xaDataSource.setUniqueResourceName("xads");
    return xaDataSource;
}

@PersistenceContext
EntityManager entityManager;

@Autowired
JmsTemplate jmsTemplate;

@Transactional(propagation = Propagation.REQUIRED)
public void process(ENTITY, MESSAGE) {
   entityManager.persist(ENTITY);
   jmsTemplate.convertAndSend(DESTINATION, MESSAGE);
}

@Autowired
private PlatformTransactionManager transactionManager;

public void process(ENTITY, MESSAGE) {
    TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
    transactionTemplate.executeWithoutResult(status -> {
        entityManager.persist(ENTITY);
        jmsTemplate.convertAndSend(DESTINATION, MESSAGE);
    });
}