【Atomikos】分布式事务简单示例

简介

基于Derby数据库，使用Atomikos框架开发一些简单的示例，包括与spring集成。

使用版本

JDK 11
atomikos 5.0.8
Spring 5.3.13

Atomikos + Derby 示例

DirectApplication.java

public class DirectApplication {private DataSource inventoryDataSource;private DataSource orderDataSource;public DirectApplication(DataSource inventoryDataSource, DataSource orderDataSource) {this.inventoryDataSource = inventoryDataSource;this.orderDataSource = orderDataSource;}public void placeOrder(String productId, int amount) throws Exception {UserTransactionImp utx = new UserTransactionImp();String orderId = UUID.randomUUID().toString();boolean rollback = false;try {utx.begin();Connection inventoryConnection = inventoryDataSource.getConnection();Connection orderConnection = orderDataSource.getConnection();Statement s1 = inventoryConnection.createStatement();String q1 = "update Inventory set balance = balance - " + amount + " where productId ='" + productId + "'";s1.executeUpdate(q1);s1.close();Statement s2 = orderConnection.createStatement();String q2 = "insert into Orders values ( '" + orderId + "', '" + productId + "', " + amount + " )";s2.executeUpdate(q2);s2.close();inventoryConnection.close();orderConnection.close();} catch (Exception e) {System.out.println(e.getMessage());rollback = true;} finally {if (!rollback)utx.commit();elseutx.rollback();}}}

说明：placeOrder方法中，分别在两个数据源中执行了SQL语句。

测试用例DirectApplicationTest.java

public class DirectApplicationTest {private static DataSource inventoryDataSource;private static DataSource orderDataSource;private static String productId = UUID.randomUUID().toString();@Testpublic void testPlaceOrderSuccess() throws Exception {int amount = 1;long initialBalance = getBalance(inventoryDataSource, productId);DirectApplication application = new DirectApplication(inventoryDataSource, orderDataSource);application.placeOrder(productId, amount);long finalBalance = getBalance(inventoryDataSource, productId);assertEquals(initialBalance - amount, finalBalance);}@Testpublic void testPlaceOrderFailure() throws Exception {int amount = 10;long initialBalance = getBalance(inventoryDataSource, productId);DirectApplication application = new DirectApplication(inventoryDataSource, orderDataSource);application.placeOrder(productId, amount);long finalBalance = getBalance(inventoryDataSource, productId);assertEquals(initialBalance, finalBalance);}@BeforeAllpublic static void setUp() throws SQLException {// 设置derby数据目录System.setProperty("derby.system.home", Paths.get("target").toAbsolutePath().toString());inventoryDataSource = getDataSource("db1");orderDataSource = getDataSource("db2");Connection inventoryConnection = inventoryDataSource.getConnection();Connection orderConnection = orderDataSource.getConnection();String createInventoryTable = "create table Inventory ( "+ " productId VARCHAR ( 100 ) PRIMARY KEY, balance INT )";String createInventoryRow = "insert into Inventory values ( '" + productId + "', 10000 )";Statement s1 = inventoryConnection.createStatement();try {s1.executeUpdate(createInventoryTable);} catch (Exception e) {System.out.println("Inventory table exists");}try {s1.executeUpdate(createInventoryRow);} catch (Exception e) {System.out.println("Product row exists");}s1.close();String createOrderTable = "create table Orders ( orderId VARCHAR ( 100 ) PRIMARY KEY, productId VARCHAR ( 100 ), amount INT NOT NULL CHECK (amount <= 5) )";Statement s2 = orderConnection.createStatement();try {s2.executeUpdate(createOrderTable);} catch (Exception e) {System.out.println("Orders table exists");}s2.close();inventoryConnection.close();orderConnection.close();}private static DataSource getDataSource(String db) {DataSource ds;AtomikosDataSourceBean ads = new AtomikosDataSourceBean();ads.setXaDataSourceClassName("org.apache.derby.jdbc.EmbeddedXADataSource");Properties properties = new Properties();properties.put("databaseName", db);properties.put("createDatabase", "create");ads.setXaProperties(properties);ads.setUniqueResourceName(db);ads.setPoolSize(10); // optionalads.setBorrowConnectionTimeout(10); // optionalds = ads;return ds;}private static long getBalance(DataSource inventoryDataSource, String productId) throws Exception {UserTransactionImp utx = new UserTransactionImp();utx.begin();Connection inventoryConnection = inventoryDataSource.getConnection();Statement s1 = inventoryConnection.createStatement();String q1 = "select balance from Inventory where productId='" + productId + "'";ResultSet rs1 = s1.executeQuery(q1);if (rs1 == null || !rs1.next())throw new Exception("Product not found: " + productId);long balance = rs1.getLong(1);inventoryConnection.close();utx.commit();return balance;}

说明：创建数据库时，Orders表的amount字段定义了规则，这个规则可能导致执行SQL语句失败。

Atomikos + Derby + Spring 示例

使用Bean方式定义数据源，代码如下：

    @Configuration@EnableTransactionManagementpublic static class Config {static {// 设置derby数据目录System.setProperty("derby.system.home", Paths.get("target").toAbsolutePath().toString());}/*** 库存 数据源* @return*/@Bean(initMethod = "init", destroyMethod = "close")public AtomikosDataSourceBean inventoryDataSource() {AtomikosDataSourceBean dataSource = new AtomikosDataSourceBean();dataSource.setLocalTransactionMode(true);dataSource.setUniqueResourceName("db1");dataSource.setXaDataSourceClassName("org.apache.derby.jdbc.EmbeddedXADataSource");Properties xaProperties = new Properties();xaProperties.put("databaseName", "db1");xaProperties.put("createDatabase", "create");dataSource.setXaProperties(xaProperties);dataSource.setPoolSize(10);return dataSource;}/*** 订单 数据源* @return*/@Bean(initMethod = "init", destroyMethod = "close")public AtomikosDataSourceBean orderDataSource() {AtomikosDataSourceBean dataSource = new AtomikosDataSourceBean();dataSource.setLocalTransactionMode(true);dataSource.setUniqueResourceName("db2");dataSource.setXaDataSourceClassName("org.apache.derby.jdbc.EmbeddedXADataSource");Properties xaProperties = new Properties();xaProperties.put("databaseName", "db2");xaProperties.put("createDatabase", "create");dataSource.setXaProperties(xaProperties);dataSource.setPoolSize(10);return dataSource;}/*** 事务管理器* * @return* @throws SystemException*/@Bean(initMethod = "init", destroyMethod = "close")public UserTransactionManager userTransactionManager() throws SystemException {UserTransactionManager userTransactionManager = new UserTransactionManager();userTransactionManager.setTransactionTimeout(300);userTransactionManager.setForceShutdown(true);return userTransactionManager;}/*** JTA事务管理器* * @return* @throws SystemException*/@Beanpublic JtaTransactionManager jtaTransactionManager() throws SystemException {JtaTransactionManager jtaTransactionManager = new JtaTransactionManager();jtaTransactionManager.setTransactionManager(userTransactionManager());jtaTransactionManager.setUserTransaction(userTransactionManager());return jtaTransactionManager;}@Beanpublic SpringApplication application() {return new SpringApplication(inventoryDataSource(), orderDataSource());}}

说明：在Spring上下文中创建了两个数据源，JTA事务管理器等实例。测试用例如下（SpringApplicationTest.java）：

@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@ExtendWith(SpringExtension.class)
@ContextConfiguration(classes = { Config.class })
public class SpringApplicationTest {private static String productId = UUID.randomUUID().toString();@AutowiredSpringApplication application;@AutowiredDataSource inventoryDataSource;@AutowiredDataSource orderDataSource;@Testpublic void testPlaceOrderSuccess() throws Exception {int amount = 1;long initialBalance = getBalance(inventoryDataSource, productId);application.placeOrder(productId, amount);long finalBalance = getBalance(inventoryDataSource, productId);assertEquals(initialBalance - amount, finalBalance);}@Testpublic void testPlaceOrderFailure() throws Exception {int amount = 10;long initialBalance = getBalance(inventoryDataSource, productId);try {application.placeOrder(productId, amount);} catch (Exception e) {System.out.println(e.getMessage());}long finalBalance = getBalance(inventoryDataSource, productId);assertEquals(initialBalance, finalBalance);}@BeforeAllpublic void setUp() throws SQLException {Connection inventoryConnection = inventoryDataSource.getConnection();Connection orderConnection = orderDataSource.getConnection();String createInventoryTable = "create table Inventory ( " + " productId VARCHAR ( 100 ) PRIMARY KEY, balance INT )";String createInventoryRow = "insert into Inventory values ( '" + productId + "', 10000 )";Statement s1 = inventoryConnection.createStatement();try {s1.executeUpdate(createInventoryTable);} catch (Exception e) {System.out.println("Inventory table exists");}try {s1.executeUpdate(createInventoryRow);} catch (Exception e) {System.out.println("Product row exists");}s1.close();String createOrderTable = "create table Orders ( orderId VARCHAR ( 100 ) PRIMARY KEY, productId VARCHAR ( 100 ), amount INT NOT NULL CHECK (amount <= 5) )";Statement s2 = orderConnection.createStatement();try {s2.executeUpdate(createOrderTable);} catch (Exception e) {System.out.println("Orders table exists");}s2.close();inventoryConnection.close();orderConnection.close();}private static long getBalance(DataSource inventoryDataSource, String productId) throws Exception {Connection inventoryConnection = inventoryDataSource.getConnection();Statement s1 = inventoryConnection.createStatement();String q1 = "select balance from Inventory where productId='" + productId + "'";ResultSet rs1 = s1.executeQuery(q1);if (rs1 == null || !rs1.next())throw new Exception("Product not found: " + productId);long balance = rs1.getLong(1);inventoryConnection.close();return balance;}

Atomikos + Derby + JPA 示例

我们需要创建实体，数据访问接口等类，这不是本文的重点，代码可以到仓库中查看。下面是库存（Inventory）配置类 InventoryConfig.java

@Configuration
@EnableJpaRepositories(basePackages = "io.github.kavahub.learnjava.jpa.inventory", entityManagerFactoryRef = "inventoryEntityManager", transactionManagerRef = "transactionManager")
public class InventoryConfig {static {// 设置derby数据目录System.setProperty("derby.system.home", Paths.get("target").toAbsolutePath().toString());}/*** 库存 数据源* * @return*/@Bean(initMethod = "init", destroyMethod = "close")public AtomikosDataSourceBean inventoryDataSource() {AtomikosDataSourceBean dataSource = new AtomikosDataSourceBean();dataSource.setLocalTransactionMode(true);dataSource.setUniqueResourceName("db1");dataSource.setXaDataSourceClassName("org.apache.derby.jdbc.EmbeddedXADataSource");Properties xaProperties = new Properties();xaProperties.put("databaseName", "db1");xaProperties.put("createDatabase", "create");dataSource.setXaProperties(xaProperties);dataSource.setPoolSize(10);return dataSource;}/*** 库存 实体管理器* * @return*/@Beanpublic EntityManagerFactory inventoryEntityManager() {HibernateJpaVendorAdapter vendorAdapter = new HibernateJpaVendorAdapter();LocalContainerEntityManagerFactoryBean factory = new LocalContainerEntityManagerFactoryBean();factory.setJpaVendorAdapter(vendorAdapter);factory.setPackagesToScan("io.github.kavahub.learnjava.jpa.inventory");factory.setDataSource(inventoryDataSource());Properties jpaProperties = new Properties();//jpaProperties.put("hibernate.show_sql", "true");//jpaProperties.put("hibernate.format_sql", "true");jpaProperties.put("hibernate.dialect", "org.hibernate.dialect.DerbyDialect");jpaProperties.put("hibernate.current_session_context_class", "jta");jpaProperties.put("javax.persistence.transactionType", "jta");jpaProperties.put("hibernate.transaction.manager_lookup_class", "com.atomikos.icatch.jta.hibernate3.TransactionManagerLookup");jpaProperties.put("hibernate.hbm2ddl.auto", "create-drop");factory.setJpaProperties(jpaProperties);factory.afterPropertiesSet();return factory.getObject();}}

订单（Order）配置类 OrderConfig.java代码如下：

@Configuration
@EnableJpaRepositories(basePackages = "io.github.kavahub.learnjava.jpa.order", entityManagerFactoryRef = "orderEntityManager", transactionManagerRef = "transactionManager")
public class OrderConfig {static {// 设置derby数据目录System.setProperty("derby.system.home", Paths.get("target").toAbsolutePath().toString());}/*** 订单 数据源* @return*/@Bean(initMethod = "init", destroyMethod = "close")public AtomikosDataSourceBean orderDataSource() {AtomikosDataSourceBean dataSource = new AtomikosDataSourceBean();dataSource.setLocalTransactionMode(true);dataSource.setUniqueResourceName("db2");dataSource.setXaDataSourceClassName("org.apache.derby.jdbc.EmbeddedXADataSource");Properties xaProperties = new Properties();xaProperties.put("databaseName", "db2");xaProperties.put("createDatabase", "create");dataSource.setXaProperties(xaProperties);dataSource.setPoolSize(10);return dataSource;}/*** 订单 实体管理器* * @return*/@Beanpublic EntityManagerFactory orderEntityManager() {HibernateJpaVendorAdapter vendorAdapter = new HibernateJpaVendorAdapter();LocalContainerEntityManagerFactoryBean factory = new LocalContainerEntityManagerFactoryBean();factory.setJpaVendorAdapter(vendorAdapter);factory.setPackagesToScan("io.github.kavahub.learnjava.jpa.order");factory.setDataSource(orderDataSource());Properties jpaProperties = new Properties();//jpaProperties.put("hibernate.show_sql", "true");//jpaProperties.put("hibernate.format_sql", "true");jpaProperties.put("hibernate.dialect", "org.hibernate.dialect.DerbyDialect");jpaProperties.put("hibernate.current_session_context_class", "jta");jpaProperties.put("javax.persistence.transactionType", "jta");jpaProperties.put("hibernate.transaction.manager_lookup_class", "com.atomikos.icatch.jta.hibernate3.TransactionManagerLookup");jpaProperties.put("hibernate.hbm2ddl.auto", "create-drop");factory.setJpaProperties(jpaProperties);factory.afterPropertiesSet();return factory.getObject();}
}

Spring主配置代码：

@Configuration@EnableTransactionManagementpublic static class Config {/*** 配置事务管理器* @return* @throws SystemException*/@Bean(initMethod = "init", destroyMethod = "close")public UserTransactionManager userTransactionManager() throws SystemException {UserTransactionManager userTransactionManager = new UserTransactionManager();userTransactionManager.setTransactionTimeout(300);userTransactionManager.setForceShutdown(true);return userTransactionManager;}/*** 配置JTA事务管理器* @return* @throws SystemException*/@Beanpublic JtaTransactionManager transactionManager() throws SystemException {JtaTransactionManager jtaTransactionManager = new JtaTransactionManager();jtaTransactionManager.setTransactionManager(userTransactionManager());jtaTransactionManager.setUserTransaction(userTransactionManager());return jtaTransactionManager;}@Beanpublic JPAApplication application() {return new JPAApplication();}}

测试用例如下：

@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@ExtendWith(SpringExtension.class)
@ContextConfiguration(classes = { Config.class, InventoryConfig.class, OrderConfig.class })
public class JPAApplicationTest {private static String productId = UUID.randomUUID().toString();@AutowiredJPAApplication application;@AutowiredInventoryRepository inventoryRepository;@AutowiredOrderRepository orderRepository;@Testpublic void testPlaceOrderSuccess() throws Exception {int amount = 1;long initialBalance = getBalance(inventoryRepository, productId);application.placeOrder(productId, amount);long finalBalance = getBalance(inventoryRepository, productId);assertEquals(initialBalance - amount, finalBalance);}@Testpublic void testPlaceOrderFailure() throws Exception {int amount = 10;long initialBalance = getBalance(inventoryRepository, productId);try {application.placeOrder(productId, amount);} catch (Exception e) {System.out.println(e.getMessage());}long finalBalance = getBalance(inventoryRepository, productId);assertEquals(initialBalance, finalBalance);}@BeforeAllpublic void setUp() throws SQLException {Inventory inventory = new Inventory();inventory.setProductId(productId);inventory.setBalance(Long.valueOf(10000));inventoryRepository.save(inventory);}private static long getBalance(InventoryRepository inventoryRepository, String productId) throws Exception {return inventoryRepository.findById(productId).orElseThrow(() -> new EntityNotFoundException(productId)).getBalance();}
}

最后

全部的代码，可以在这里查看 atomikos欢迎顶赞，感谢！

【Atomikos】分布式事务简单示例相关推荐

jta分布式事务简单使用
个人理解总结: jta分布式事务,其实与普通的spring事务配置没多大区别,最主要的是配置不同的mapper使用不同的的数据源,然后再配atomikos和spring的jta事务管理器进行管理(sp ...
atomikos mysql,记一次 Atomikos 分布式事务的使用
由于项目上的需要,我要同时往orcale数据库与sqlserver数据中插入数据,需要在一个事务之内完成这两个库的提交.参考了一下网上的各种JTA(Java Transaction API)实现之后, ...
破解世界性技术难题！ GTS让分布式事务简单高效
近日,2017云栖大会·深圳峰会如期举行,多项阿里云新产品对外发布.在企业级互联网架构分会场,来自阿里中间件(Aliware)的技术专家及合作伙伴,为现场参会嘉宾带来最新的传统IT架构到企业级互联网架 ...
Atomikos 分布式事务的使用
由于项目上的需要,我要同时往orcale数据库与sqlserver数据中插入数据,需要在一个事务之内完成这两个库的提交.参考了一下网上的各种JTA(Java Transaction API)实现之后, ...
mysql5.6 分布式事务_mysql 分布式事务xa 示例
/***XA MYSQL**/ //连接数据库1 function getConnect1() { $con = mysql_connect("localhost","r ...
使用atomikos分布式事务报com.atomikos.icatch.RollbackException: Prepare: NO vote异常解决办法
发现后台使用atomikos进行事务提交时报javax.transaction.RollbackException: Prepare: NO vote,造成这个事务的主要原因是使用atomikos时, ...
分布式事务解决方案——基于Atomikos的实现
声明:以下关于"JTA规范事务模型"."Spring JTA分布式事务的实现"等内容均来源于其他大佬的博客内容,并已经表明出处. 1.JTA规范事务模型 J ...
分布式事务（含面试题）- 图解 - 秒懂 - 史上最全
文章很长,而且持续更新,建议收藏起来,慢慢读! Java 高并发发烧友社群:疯狂创客圈(总入口) 奉上以下珍贵的学习资源: 免费赠送经典图书 : 极致经典 + 社群大片好评 < Java 高 ...
分布式事务（秒懂）
文章很长,而且持续更新,建议收藏起来,慢慢读! Java 高并发发烧友社群:疯狂创客圈(总入口) 奉上以下珍贵的学习资源: 免费赠送经典图书 : 极致经典 + 社群大片好评 < Java 高 ...
SpringCloud Alibaba 2021微服务实战三十二集成RocketMQ实现分布式事务
目录基于RocketMQ分布式事务 - 完整示例 2.解决方案 2.1.本地消息表方案 2.2.RocketMQ事务消息方案一.事务消息二.订单服务 1.事务日志表 2.TransactionM ...

【Atomikos】分布式事务简单示例

简介

Atomikos + Derby 示例

Atomikos + Derby + Spring 示例

Atomikos + Derby + JPA 示例

最后

【Atomikos】分布式事务简单示例相关推荐

最新文章

热门文章