`
还可以
  • 浏览: 80023 次
  • 性别: Icon_minigender_1
  • 来自: 重庆
社区版块
存档分类
最新评论

JTA分布式事务实战(atomikos)

jtaspring3.0-jtaatomikos分布式事务 
阅读更多

最近需要用到分布式事务，研究了下jta，使用了atomikos这个jta的实现，使用的是spring3.0，废话少说，直接贴代码。

1.使用如下jar包

atomikos-util.3.7.0.jar

cglib-nodep-2.2.2.jar

transactions-3.7.0.jar

transactions-api-3.7.0.jar

transactions-jdbc-3.7.0.jar

transactions-jta-3.7.0.jar

 

2.spring配置文件如下:

   

<?xml version="1.0" encoding="UTF-8"?>  
<beans xmlns="http://www.springframework.org/schema/beans"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jee="http://www.springframework.org/schema/jee"  
    xmlns:tx="http://www.springframework.org/schema/tx" xmlns:context="http://www.springframework.org/schema/context"  
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-3.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd"  
    default-lazy-init="true">  
  
  	
  	<context:component-scan base-package="com.atom.jta.test" />
  
    <!-- atomikos事务管理器 -->  
    <bean id="atomikosTransactionManager" class="com.atomikos.icatch.jta.UserTransactionManager"  
        init-method="init" destroy-method="close">  
        <description>UserTransactionManager</description>  
        <property name="forceShutdown">  
            <value>true</value>  
        </property>  
    </bean>  
  
    <bean id="atomikosUserTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">  
        <property name="transactionTimeout" value="300" />  
    </bean>  
  
    <!-- spring 事务管理器 -->  
    <bean id="springTransactionManager"  
        class="org.springframework.transaction.jta.JtaTransactionManager">  
        <property name="transactionManager">  
            <ref bean="atomikosTransactionManager" />  
        </property>  
        <property name="userTransaction">  
            <ref bean="atomikosUserTransaction" />  
        </property>  
        <property name="allowCustomIsolationLevels" value="true">
        </property>
    </bean>  
    
    <!-- 事务拦截器 -->
	<bean id="transactionInterceptor"
		class="org.springframework.transaction.interceptor.TransactionInterceptor">
		<!-- 配置事务管理器 -->
		<property name="transactionManager" ref="springTransactionManager" />
		<!-- 方法名：要求的事务属性 -->
		<property name="transactionAttributes">
			<props>
				<prop key="insertTest">PROPAGATION_REQUIRED,ISOLATION_REPEATABLE_READ,-Throwable
				</prop>
			</props>
		</property>
	</bean>
	
	<bean
		class="org.springframework.aop.framework.autoproxy.BeanNameAutoProxyCreator">
		<property name="beanNames">
			<list>
				<value>atomTransactionServiceImpl</value>
			</list>
		</property>
		<property name="interceptorNames">
			<list>
				<value>transactionInterceptor</value>
			</list>
		</property>
	</bean>
</beans>

 

 

2.数据源:

datasource1:

 

package com.atom.jta.test;

import java.util.Properties;

import org.springframework.stereotype.Repository;

import com.atomikos.jdbc.AtomikosDataSourceBean;
@Repository
public class MasterAtomDatasource extends AtomikosDataSourceBean {
	
	private static final long serialVersionUID = -2471230875536339311L;

	public MasterAtomDatasource(){
		Properties prop = new Properties();
		prop.put("user", "root");
		prop.put("password", "");
		prop.put("URL", "jdbc:mysql://127.0.0.1:3306/test?autoReconnect=true");
		setXaDataSourceClassName("com.mysql.jdbc.jdbc2.optional.MysqlXADataSource");
		setUniqueResourceName("mysql_ds1");
		setPoolSize(5);
		setXaProperties(prop);
	}
	
}

 datasource2:

 

 

package com.atom.jta.test;

import java.util.Properties;

import org.springframework.stereotype.Repository;

import com.atomikos.jdbc.AtomikosDataSourceBean;
@Repository
public class SlaveAtomDataSource extends AtomikosDataSourceBean {

	private static final long serialVersionUID = -6210394799199416765L;
	public SlaveAtomDataSource(){
		Properties prop = new Properties();
		prop.put("user", "root");
		prop.put("password", "");
		prop.put("URL", "jdbc:mysql://127.0.0.1:3306/test1?autoReconnect=true");
		setXaDataSourceClassName("com.mysql.jdbc.jdbc2.optional.MysqlXADataSource");
		setUniqueResourceName("mysql_ds2");
		setPoolSize(5);
		setXaProperties(prop);
	}
	

}

 

 

3.BaseDao:

package com.atom.jta.test;

import javax.annotation.Resource;
import javax.sql.DataSource;

import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;
@Repository
public class AtomBaseDao {
	private JdbcTemplate mastTemplate;
	private JdbcTemplate slaveTemplate;
	
	public JdbcTemplate getMastTemplate() {
		return mastTemplate;
	}
	@Resource(name="masterAtomDatasource")
	public void setMastTemplate(DataSource source) {
		this.mastTemplate = new JdbcTemplate(source);
	}
	public JdbcTemplate getSlaveTemplate() {
		return slaveTemplate;
	}
	@Resource(name="slaveAtomDataSource")
	public void setSlaveTemplate(DataSource source) {
		this.slaveTemplate = new JdbcTemplate(source);
	}
	
	
}

 4.测试service

package com.atom.jta.test;

import org.springframework.context.ApplicationContext;

public interface AtomTransactionService {
	public void insertTest(ApplicationContext ctx) throws Exception;
}

 

package com.atom.jta.test;

import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;
@Service
public class AtomTransactionServiceImpl implements AtomTransactionService {
	public void insertTest(ApplicationContext ctx) throws Exception {
		AtomBaseDao baseDao = ctx.getBean(AtomBaseDao.class);
		String str = "xxxx";
		String masterSql = "insert into demo (name) values "+"('"+str+"')";
		String slaveSql = "insert into test (name) values "+"('"+str+"')";
		baseDao.getMastTemplate().execute(masterSql);
		baseDao.getSlaveTemplate().execute(slaveSql);
		throw new Exception();
	}

}

 5.事务测试

package com.atom.jta.test;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class AtomTest {
	public AtomTest(){
		ApplicationContext ctx = new ClassPathXmlApplicationContext("applicationContext-atomikos.xml");
		AtomTransactionService service = ctx.getBean(AtomTransactionService.class);
		try {
			service.insertTest(ctx);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public static void main(String[] args) {
		AtomTest test = new AtomTest();
		System.out.println("done.....");
	}

}

 就是如此简单

分享到：
| mysql主备复制搭建
评论
6 楼 小黄牛 2016-08-10  
之前有看过阿里的TCP，看到他们的架构图中可以通过自有业务去控制MQ中的消息的状态以实现消息是否能够被消费者消费。了解了下ActiveMQ,是不支持修改MQ内消息的状态的，其他的消息中间件就不知道是否支持了。
不过，现在的分布式事务的处理也有很多方案，比如可靠消息最终一致性、TCC事务补偿型、进最大努力通知型。方案多多，就看你怎么实现。有兴趣你也可以看下这篇博客，它里面的实现就有点阿里的TCP实现，包含消息状态处理、消息确认什么的。http://www.roncoo.com/article/detail/124243
5 楼 leecyz 2016-01-03  
还可以 写道
lvwenwen 写道
哥们，能上传工程吗

不要用这个，性能太差。我们现在的解决方案是做了事务拆分，将事务分为本地事务和远端事务，远端事务通过异步去做，各自保证事务一定成功。


的确是这样，xts的实现原理，最终一致性，并且也很好理解。
jta的2pc资料好少，没什么人讲下实现原理的，现在还不清楚具体怎么实现的。
4 楼 还可以 2014-04-18  
lvwenwen 写道
哥们，能上传工程吗

不要用这个，性能太差。我们现在的解决方案是做了事务拆分，将事务分为本地事务和远端事务，远端事务通过异步去做，各自保证事务一定成功。
3 楼 还可以 2014-04-18  
heipacker 写道
“最近需要用到分布式事务，研究了下jta，使用了atomikos这个jta的实现，使用的是spring3.0，废话少说，直接贴代码。”关键是你能说吗除了代码
这东西在生产环境性能太差了，做性能测试的时候并发完全压不上去，高并发的系统千万不要用。
2 楼 heipacker 2014-03-31  
“最近需要用到分布式事务，研究了下jta，使用了atomikos这个jta的实现，使用的是spring3.0，废话少说，直接贴代码。”关键是你能说吗除了代码
1 楼 lvwenwen 2014-03-28  
哥们，能上传工程吗
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

Magicbox
Global site tag (gtag.js) - Google Analytics