使用Java开源组件Atomikos开发分布式事务应用

2024-04-24 14:08

本文主要是介绍使用Java开源组件Atomikos开发分布式事务应用,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

Atomikos是一个公司的名字,AtomikosTransactionsEssentials是其开源的分布式事务软件包,而ExtremeTransactions是商业的分布式事务软件包。TransactionsEssentials是基于apache-license的,是JTA/XA的开源实现,支持Java Application和J2EE应用。
下面以AtomikosTransactionsEssentials-3.4.2(可以在 http://www.atomikos.com下载)为例,说明其用法。
需要的jar包:jta.jar、transactions-essentials-all.jar。
Atomikos默认在classpath下使用名为transactions.properties的配置文件,如果找不到,则使用默认的配置参数。下面给一个transactions.properties的例子,可以根据自己的需要修改:
#SAMPLE PROPERTIES FILE FOR THE TRANSACTION SERVICE
#THIS FILE ILLUSTRATES THE DIFFERENT SETTINGS FOR THE TRANSACTION MANAGER
#UNCOMMENT THE ASSIGNMENTS TO OVERRIDE DEFAULT VALUES;
#Required: factory class name for the transaction service core.
#
com.atomikos.icatch.service=com.atomikos.icatch.standalone.UserTransactionServiceFactory
#
#Set name of file where messages are output
#
#com.atomikos.icatch.console_file_name = tm.out
#Size limit (in bytes) for the console file;
#negative means unlimited.
#
#com.atomikos.icatch.console_file_limit=-1
#For size-limited console files, this option
#specifies a number of rotating files to
#maintain.
#
#com.atomikos.icatch.console_file_count=1
#Set the number of log writes between checkpoints
#
#com.atomikos.icatch.checkpoint_interval=500
#Set output directory where console file and other files are to be put
#make sure this directory exists!
#
#com.atomikos.icatch.output_dir = ./
#Set directory of log files; make sure this directory exists!
#
#com.atomikos.icatch.log_base_dir = ./
#Set base name of log file
#this name will be used as the first part of
#the system-generated log file name
#
#com.atomikos.icatch.log_base_name = tmlog
#Set the max number of active local transactions
#or -1 for unlimited.
#
#com.atomikos.icatch.max_actives = 50
#Set the max timeout (in milliseconds) for local transactions
#
#com.atomikos.icatch.max_timeout = 300000
#The globally unique name of this transaction manager process
#override this value with a globally unique name
#
#com.atomikos.icatch.tm_unique_name = tm
#Do we want to use parallel subtransactions? JTA's default
#is NO for J2EE compatibility.
#
#com.atomikos.icatch.serial_jta_transactions=true
#If you want to do explicit resource registration then
#you need to set this value to false. See later in
#this manual for what explicit resource registration means.
#
#com.atomikos.icatch.automatic_resource_registration=true
#Set this to WARN, INFO or DEBUG to control the granularity
#of output to the console file.
#
#com.atomikos.icatch.console_log_level=WARN
#Do you want transaction logging to be enabled or not?
#If set to false, then no logging overhead will be done
#at the risk of losing data after restart or crash.
#
#com.atomikos.icatch.enable_logging=true
#Should two-phase commit be done in (multi-)threaded mode or not?
#
#com.atomikos.icatch.threaded_2pc=true
#Should exit of the VM force shutdown of the transaction core?
#
#com.atomikos.icatch.force_shutdown_on_vm_exit=false
#Should the logs be protected by a .lck file on startup?
#
#com.atomikos.icatch.lock_logs=true
Atomikos TransactionsEssentials支持3种使用方式,可以根据自己的情况选用,下面给出每种方式的使用场合和一个代码示例。
一、使用JDBC/JMS和UserTransaction,这是最直接和最简单的使用方式,使用Atomikos内置的JDBC、JMS适配器。示例如下:
package demo.atomikos;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;

import javax.transaction.UserTransaction;

import com.atomikos.icatch.jta.UserTransactionImp;
import com.atomikos.jdbc.AtomikosDataSourceBean;

/**
*
*/
public class UserTransactionUtil {

public static UserTransaction getUserTransaction() {
UserTransaction utx = new UserTransactionImp();
return utx;
}

private static AtomikosDataSourceBean dsBean;

private static AtomikosDataSourceBean getDataSource() {
if (dsBean != null) return dsBean;
AtomikosDataSourceBean ds = new AtomikosDataSourceBean();
ds.setUniqueResourceName("db");
ds.setXaDataSourceClassName("oracle.jdbc.xa.client.OracleXADataSource");
Properties p = new Properties();
p.setProperty("user", "db_user_name" );
p.setProperty("password", "db_user_pwd");
p.setProperty("URL", "jdbc:oracle:thin:@192.168.0.10:1521:oradb");
ds.setXaProperties(p);
ds.setPoolSize(5);
dsBean = ds;
return dsBean;
}

public static Connection getDbConnection() throws SQLException{
Connection conn = getDataSource().getConnection();
return conn;
}

private static AtomikosDataSourceBean dsBean1;

private static AtomikosDataSourceBean getDataSource1() {
if (dsBean1 != null) return dsBean1;
AtomikosDataSourceBean ds = new AtomikosDataSourceBean();
ds.setUniqueResourceName("db1");
ds.setXaDataSourceClassName("oracle.jdbc.xa.client.OracleXADataSource");
Properties p = new Properties();
p.setProperty("user", "db_user_name" );
p.setProperty("password", "db_user_pwd");
p.setProperty("URL", "jdbc:oracle:thin:@192.168.0.11:1521:oradb1");
ds.setXaProperties(p);
ds.setPoolSize(5);
dsBean1 = ds;
return dsBean1;
}

public static Connection getDb1Connection() throws SQLException{
Connection conn = getDataSource1().getConnection();
return conn;
}

public static void main(String[] args) {
UserTransaction utx = getUserTransaction();
boolean rollback = false;
try {
//begin a transaction
utx.begin();

//execute db operation
Connection conn = null;
Connection conn1 = null;
Statement stmt = null;
Statement stmt1 = null;
try {
conn = getDbConnection();
conn1 = getDb1Connection();

stmt = conn.createStatement();
stmt.executeUpdate("insert into t values(1,'23')");

stmt1 = conn1.createStatement();
stmt1.executeUpdate("insert into t values(1,'123456789')");

}
catch(Exception e) {
throw e;
}
finally {
if (stmt != null) stmt.close();
if (conn != null) conn.close();
if (stmt1 != null) stmt1.close();
if (conn1 != null) conn1.close();
}
}
catch(Exception e) {
//an exception means we should not commit
rollback = true;
e.printStackTrace();
}
finally {
try {
//commit or rollback the transaction
if ( !rollback ) utx.commit();
else utx.rollback();
}
catch(Exception e) {
e.printStackTrace();
}
}
}
}
二、使用JTA TransactionManager。这种方式不需要Atomikos内置的JDBC、JMS适配器,但需要在JTA/XA级别上添加、删除XA资源实例。示例如下:
package demo.atomikos;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;

import javax.sql.XAConnection;
import javax.transaction.Transaction;
import javax.transaction.xa.XAResource;

import oracle.jdbc.xa.client.OracleXADataSource;

import com.atomikos.icatch.jta.UserTransactionManager;

public class TransactionManagerUtil {
public static UserTransactionManager getUserTransactionManager() throws Exception {
return new UserTransactionManager();
}

private static OracleXADataSource xads;

private static OracleXADataSource getXADataSource() throws SQLException{
if (xads != null) return xads;
xads = new OracleXADataSource();
xads.setUser ("db_user_name");
xads.setPassword("db_user_pwd");
xads.setURL("jdbc:oracle:thin:@192.168.0.10:1521:oradb");
return xads;
}

public static XAConnection getXAConnection() throws SQLException{
OracleXADataSource ds = getXADataSource();
return ds.getXAConnection();
}

private static OracleXADataSource xads1;

private static OracleXADataSource getXADataSource1() throws SQLException{
if (xads1 != null) return xads1;
xads1 = new OracleXADataSource();
xads1.setUser ("db_user_name");
xads1.setPassword("db_user_pwd");
xads1.setURL("jdbc:oracle:thin:@192.168.0.11:1521:oradb1");
return xads1;
}

public static XAConnection getXAConnection1() throws SQLException{
OracleXADataSource ds = getXADataSource1();
return ds.getXAConnection();
}

public static void main(String[] args) {
try {
UserTransactionManager tm = getUserTransactionManager();

XAConnection xaconn = getXAConnection();
XAConnection xaconn1 = getXAConnection1();

boolean rollback = false;
try {
//begin and retrieve tx
tm.begin();
Transaction tx = tm.getTransaction();

//get the XAResourc from the JDBC connection
XAResource xares = xaconn.getXAResource();
XAResource xares1 = xaconn1.getXAResource();

//enlist the resource with the transaction
//NOTE: this will only work if you set the configuration parameter:
//com.atomikos.icatch.automatic_resource_registration=true
//or, alternatively, if you use the UserTransactionService
//integration mode
tx.enlistResource(xares);
tx.enlistResource(xares1);

//access the database, the work will be
//subject to the outcome of the current transaction
Connection conn = xaconn.getConnection();
Statement stmt = conn.createStatement();
stmt.executeUpdate("insert into t values(1,'1234567')");
stmt.close();
conn.close();
Connection conn1 = xaconn1.getConnection();
Statement stmt1 = conn1.createStatement();
stmt1.executeUpdate("insert into t values(1,'abc1234567890')");
stmt1.close();
conn1.close();

//delist the resource
tx.delistResource(xares, XAResource.TMSUCCESS);
tx.delistResource(xares1, XAResource.TMSUCCESS);
}
catch ( Exception e ) {
//an exception means we should not commit
rollback = true;
throw e;
}
finally {
//ALWAYS terminate the tx
if (rollback) tm.rollback();
else tm.commit();

//only now close the connection
//i.e., not until AFTER commit or rollback!
xaconn.close();
xaconn1.close();
}
}
catch (Exception e) {
e.printStackTrace();
}
}
}
三、使用Atomikos UserTransactionService。这是高级使用方式,可以控制事务服务的启动和关闭,并且可以控制资源的装配。示例如下:
package demo.atomikos;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;

import javax.sql.XAConnection;
import javax.transaction.Transaction;
import javax.transaction.TransactionManager;
import javax.transaction.xa.XAResource;

import oracle.jdbc.xa.client.OracleXADataSource;

import com.atomikos.datasource.xa.jdbc.JdbcTransactionalResource;
import com.atomikos.icatch.config.TSInitInfo;
import com.atomikos.icatch.config.UserTransactionService;
import com.atomikos.icatch.config.UserTransactionServiceImp;

public class UserTransactionServiceUtil {
public static UserTransactionService getUserTransactionService() throws Exception {
return new UserTransactionServiceImp();
}

private static OracleXADataSource xads;

private static OracleXADataSource getXADataSource() throws SQLException{
if (xads != null) return xads;
xads = new OracleXADataSource();
xads.setUser ("db_user_name");
xads.setPassword("db_user_pwd");
xads.setURL("jdbc:oracle:thin:@192.168.0.10:1521:oradb");
return xads;
}

public static XAConnection getXAConnection() throws SQLException{
OracleXADataSource ds = getXADataSource();
return ds.getXAConnection();
}

private static JdbcTransactionalResource jdbcResource;

public static JdbcTransactionalResource getJdbcTransactionalResource() throws SQLException{
if (jdbcResource != null) return jdbcResource;
jdbcResource = new JdbcTransactionalResource (
"db"
,getXADataSource()
,new com.atomikos.datasource.xa.OraXidFactory() //oracle db need this
);
return jdbcResource;
}

private static OracleXADataSource xads1;

private static OracleXADataSource getXADataSource1() throws SQLException{
if (xads1 != null) return xads1;
xads1 = new OracleXADataSource();
xads1.setUser ("db_user_name");
xads1.setPassword("db_user_pwd");
xads1.setURL("jdbc:oracle:thin:@192.168.0.11:1521:oradb1");
return xads1;
}

public static XAConnection getXAConnection1() throws SQLException{
OracleXADataSource ds = getXADataSource1();
return ds.getXAConnection();
}

private static JdbcTransactionalResource jdbcResource1;

public static JdbcTransactionalResource getJdbcTransactionalResource1() throws SQLException{
if (jdbcResource1 != null) return jdbcResource1;
jdbcResource1 = new JdbcTransactionalResource (
"db1"
,getXADataSource1()
,new com.atomikos.datasource.xa.OraXidFactory() //oracle db need this
);
return jdbcResource1;
}

public static void main(String[] args) {
try {

//Register the resource with the transaction service
//this is done through the UserTransaction handle.
//All UserTransaction instances are equivalent and each
//one can be used to register a resource at any time.
UserTransactionService uts = getUserTransactionService();
uts.registerResource(getJdbcTransactionalResource());
uts.registerResource(getJdbcTransactionalResource1());

//Initialize the UserTransactionService.
//This will start the TM and recover
//all registered resources; you could
//call this 'eager recovery' (as opposed to 'lazy recovery'
//for the simple xa demo).
TSInitInfo info = uts.createTSInitInfo();
//optionally set config properties on info
info.setProperty("com.atomikos.icatch.checkpoint_interval", "2000");
uts.init(info);

TransactionManager tm = uts.getTransactionManager();
tm.setTransactionTimeout(60);

XAConnection xaconn = getXAConnection();
XAConnection xaconn1 = getXAConnection1();

boolean rollback = false;

//begin and retrieve tx
tm.begin();
Transaction tx = tm.getTransaction();

//get the XAResourc from the JDBC connection
XAResource xares = xaconn.getXAResource();
XAResource xares1 = xaconn1.getXAResource();

Connection conn = xaconn.getConnection();
Connection conn1 = xaconn1.getConnection();
try {

//enlist the resource with the transaction
//NOTE: this will only work if you set the configuration parameter:
//com.atomikos.icatch.automatic_resource_registration=true
//or, alternatively, if you use the UserTransactionService
//integration mode
tx.enlistResource(xares);
tx.enlistResource(xares1);

//access the database, the work will be
//subject to the outcome of the current transaction
Statement stmt = conn.createStatement();
stmt.executeUpdate("insert into t values(1,'1234567')");
stmt.close();

Statement stmt1 = conn1.createStatement();
stmt1.executeUpdate("insert into t values(1,'abc')");
stmt1.close();

}
catch ( Exception e ) {
//an exception means we should not commit
rollback = true;
throw e;
}
finally {
int flag = XAResource.TMSUCCESS;
if (rollback) flag = XAResource.TMFAIL;

tx.delistResource(xares, flag);
tx.delistResource(xares1, flag);

conn.close();
conn1.close();

if (!rollback) tm.commit();
else tm.rollback();
}

uts.shutdown(false);
}
catch (Exception e) {
e.printStackTrace();
}
}
}
在使用的时候,也可以把资源等配置到应用服务器中,使用JNDI获取资源。也可以与spring集成。

这篇关于使用Java开源组件Atomikos开发分布式事务应用的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/931974

相关文章

springboot集成easypoi导出word换行处理过程

《springboot集成easypoi导出word换行处理过程》SpringBoot集成Easypoi导出Word时,换行符n失效显示为空格,解决方法包括生成段落或替换模板中n为回车,同时需确... 目录项目场景问题描述解决方案第一种:生成段落的方式第二种:替换模板的情况,换行符替换成回车总结项目场景s

SpringBoot集成redisson实现延时队列教程

《SpringBoot集成redisson实现延时队列教程》文章介绍了使用Redisson实现延迟队列的完整步骤,包括依赖导入、Redis配置、工具类封装、业务枚举定义、执行器实现、Bean创建、消费... 目录1、先给项目导入Redisson依赖2、配置redis3、创建 RedissonConfig 配

SpringBoot中@Value注入静态变量方式

《SpringBoot中@Value注入静态变量方式》SpringBoot中静态变量无法直接用@Value注入,需通过setter方法,@Value(${})从属性文件获取值,@Value(#{})用... 目录项目场景解决方案注解说明1、@Value("${}")使用示例2、@Value("#{}"php

SpringBoot分段处理List集合多线程批量插入数据方式

《SpringBoot分段处理List集合多线程批量插入数据方式》文章介绍如何处理大数据量List批量插入数据库的优化方案:通过拆分List并分配独立线程处理,结合Spring线程池与异步方法提升效率... 目录项目场景解决方案1.实体类2.Mapper3.spring容器注入线程池bejsan对象4.创建

线上Java OOM问题定位与解决方案超详细解析

《线上JavaOOM问题定位与解决方案超详细解析》OOM是JVM抛出的错误,表示内存分配失败,:本文主要介绍线上JavaOOM问题定位与解决方案的相关资料,文中通过代码介绍的非常详细,需要的朋... 目录一、OOM问题核心认知1.1 OOM定义与技术定位1.2 OOM常见类型及技术特征二、OOM问题定位工具

基于 Cursor 开发 Spring Boot 项目详细攻略

《基于Cursor开发SpringBoot项目详细攻略》Cursor是集成GPT4、Claude3.5等LLM的VSCode类AI编程工具,支持SpringBoot项目开发全流程,涵盖环境配... 目录cursor是什么?基于 Cursor 开发 Spring Boot 项目完整指南1. 环境准备2. 创建

Python使用FastAPI实现大文件分片上传与断点续传功能

《Python使用FastAPI实现大文件分片上传与断点续传功能》大文件直传常遇到超时、网络抖动失败、失败后只能重传的问题,分片上传+断点续传可以把大文件拆成若干小块逐个上传,并在中断后从已完成分片继... 目录一、接口设计二、服务端实现(FastAPI)2.1 运行环境2.2 目录结构建议2.3 serv

Spring Security简介、使用与最佳实践

《SpringSecurity简介、使用与最佳实践》SpringSecurity是一个能够为基于Spring的企业应用系统提供声明式的安全访问控制解决方案的安全框架,本文给大家介绍SpringSec... 目录一、如何理解 Spring Security?—— 核心思想二、如何在 Java 项目中使用?——

SpringBoot+RustFS 实现文件切片极速上传的实例代码

《SpringBoot+RustFS实现文件切片极速上传的实例代码》本文介绍利用SpringBoot和RustFS构建高性能文件切片上传系统,实现大文件秒传、断点续传和分片上传等功能,具有一定的参考... 目录一、为什么选择 RustFS + SpringBoot?二、环境准备与部署2.1 安装 RustF

springboot中使用okhttp3的小结

《springboot中使用okhttp3的小结》OkHttp3是一个JavaHTTP客户端,可以处理各种请求类型,比如GET、POST、PUT等,并且支持高效的HTTP连接池、请求和响应缓存、以及异... 在 Spring Boot 项目中使用 OkHttp3 进行 HTTP 请求是一个高效且流行的方式。