ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 应用开发 > IT综合 > Spring 4+atomikos+JdbcTemplate+DynamicDataSource 分布式事务数据源动态切换

Spring 4+atomikos+JdbcTemplate+DynamicDataSource 分布式事务数据源动态切换

原创 IT综合 作者：百联达 时间：2017-04-12 14:38:57 0 删除 编辑
一：依赖的jar包  Maven配置


点击(此处)折叠或打开

  1. <dependency>
  2.             <groupId>com.atomikos</groupId>
  3.             <artifactId>transactions</artifactId>
  4.             <version>4.0.4</version>
  5.         </dependency>
  6.         <dependency>
  7.             <groupId>com.atomikos</groupId>
  8.             <artifactId>transactions-api</artifactId>
  9.             <version>4.0.4</version>
  10.         </dependency>
  11.         <dependency>
  12.             <groupId>com.atomikos</groupId>
  13.             <artifactId>atomikos-util</artifactId>
  14.             <version>4.0.4</version>
  15.         </dependency>
  16.         <dependency>
  17.             <groupId>com.atomikos</groupId>
  18.             <artifactId>transactions-jdbc-deprecated</artifactId>
  19.             <version>3.8.0</version>
  20.         </dependency>
  21.         <dependency>
  22.             <groupId>com.atomikos</groupId>
  23.             <artifactId>transactions-jta</artifactId>
  24.             <version>4.0.4</version>
  25.         </dependency>
  26.         <dependency>
  27.             <groupId>com.atomikos</groupId>
  28.             <artifactId>transactions-jdbc</artifactId>
  29.             <version>4.0.4</version>
  30.         </dependency>

  32.         <dependency>
  33.             <groupId>cglib</groupId>
  34.             <artifactId>cglib-nodep</artifactId>
  35.             <version>3.2.5</version>
  36.         </dependency>

  38.         <dependency>
  39.             <groupId>javax.transaction</groupId>
  40.             <artifactId>jta</artifactId>
  41.             <version>1.1</version>
  42.         </dependency>
二：配置

dataSource.properties

点击(此处)折叠或打开

  1. workDesk.jdbc.driverclass=com.mysql.jdbc.Driver
  2. workDesk.jdbc.url=jdbc:mysql://10.243.3.18:3306/system?userUnicode=true&characterEncoding=UTF-8
  3. workDesk.jdbc.username=root
  4. workDesk.jdbc.password=$Fortune2015
  5. workDesk.jdbc.poolsize.max=3
  6. workDesk.jdbc.poolsize.min=3
  7. workDesk.jdbc.poolsize.initial=2
  8. workDesk.jdbc.idletime.max=25000
  9. workDesk.jdbc.idleConnectionTestPeriod=18000

  11. #-------workDesk jdbc--------
  12. workDesk.read.jdbc.driverclass=com.mysql.jdbc.Driver
  13. workDesk.read.jdbc.url=jdbc:mysql://112.74.53.213:3306/gmc?userUnicode=true&characterEncoding=UTF-8
  14. workDesk.read.jdbc.username=root
  15. workDesk.read.jdbc.password=Wanmide@123
  16. workDesk.read.jdbc.poolsize.max=3
  17. workDesk.read.jdbc.poolsize.min=3
  18. workDesk.read.jdbc.poolsize.initial=2
  19. workDesk.read.jdbc.idletime.max=25000
  20. workDesk.read.jdbc.idleConnectionTestPeriod=18000

  22. jdbc.xaDataSourceClassName=com.mysql.jdbc.jdbc2.optional.MysqlXADataSource
transactions.properties


点击(此处)折叠或打开

  1. # SAMPLE PROPERTIES FILE FOR THE TRANSACTION SERVICE
  2. # THIS FILE ILLUSTRATES THE DIFFERENT SETTINGS FOR THE TRANSACTION MANAGER
  3. # UNCOMMENT THE ASSIGNMENTS TO OVERRIDE DEFAULT VALUES;

  5. # Required: factory implementation class of the transaction core.
  6. # NOTE: there is no default for this, so it MUST be
  7. #
  8. com.atomikos.icatch.service=com.atomikos.icatch.standalone.UserTransactionServiceFactory

  10.     
  11. # Set base name of file where messages are output
  12. # (also known as the 'console file').
  13. #
  14. # com.atomikos.icatch.console_file_name = tm.out

  16. # Size limit (in bytes) for the console file;
  17. # negative means unlimited.
  18. #
  19. # com.atomikos.icatch.console_file_limit=-1

  21. # For size-limited console files, this option
  22. # specifies a number of rotating files to
  23. # maintain.
  24. #
  25. # com.atomikos.icatch.console_file_count=1

  27. # Set the number of log writes between checkpoints
  28. #
  29. # com.atomikos.icatch.checkpoint_interval=500

  31. # Set output directory where console file and other files are to be put
  32. # make sure this directory
  33. #
  34. # com.atomikos.icatch.output_dir = ./

  36. # Set directory of log files; make sure this directory
  37. #
  38.  com.atomikos.icatch.log_base_dir = ./

  40. # Set base name of log file
  41. # this name will be used as the first part of
  42. # the system-generated log file name
  43. #
  44. # com.atomikos.icatch.log_base_name = tmlog

  46. # Set the max number of active local transactions
  47. # or -1 for unlimited.
  48. #
  49. # com.atomikos.icatch.max_actives = 50

  51. # Set the default timeout (in milliseconds) for local transactions
  52. #
  53. # com.atomikos.icatch.default_jta_timeout = 10000

  55. # Set the max timeout (in milliseconds) for local transactions
  56. #
  57. # com.atomikos.icatch.max_timeout = 300000

  59. # The globally unique name of this transaction manager process
  60. # override this value with a globally unique name
  61. #
  62. # com.atomikos.icatch.tm_unique_name = tm

  64. # Do we want to use parallel subtransactions? JTA




点击(此处)折叠或打开

  1. /**
  2.      * Atomikos 数据源A
  3.      * @return
  4.      */
  5.     @Bean(name="dataSourceA",initMethod="init",destroyMethod="close")
  6.     public AtomikosDataSourceBean dataSourceA()
  7.     {
  8.         AtomikosDataSourceBean dataSourceA=new AtomikosDataSourceBean();
  9.         dataSourceA.setUniqueResourceName("dataSourceA");
  10.         dataSourceA.setXaDataSourceClassName(xaDataSourceClassName);
  11.         Properties xaProperties = new Properties();
  12.         xaProperties.put("user", user);
  13.         xaProperties.put("password", password);
  14.         xaProperties.put("url", jdbcUrl);
  15.         xaProperties.put("pinGlobalTxToPhysicalConnection", true);
  16.         dataSourceA.setXaProperties(xaProperties);
  17.         dataSourceA.setMaxPoolSize(maxPoolSize);
  18.         dataSourceA.setMinPoolSize(minPoolSize);
  19.         dataSourceA.setMaxIdleTime(maxIdleTime);
  20.         dataSourceA.setTestQuery("SELECT 1");
  21.         return dataSourceA;
  22.     }
  23.     
  24.     
  25.     /**
  26.      * Atomikos 数据源A
  27.      * @return
  28.      */
  29.     @Bean(name="dataSourceB",initMethod="init",destroyMethod="close")
  30.     public AtomikosDataSourceBean dataSourceB()
  31.     {
  32.         AtomikosDataSourceBean dataSourceA=new AtomikosDataSourceBean();
  33.         dataSourceA.setUniqueResourceName("dataSourceB");
  34.         dataSourceA.setXaDataSourceClassName(xaDataSourceClassName);
  35.         Properties xaProperties = new Properties();
  36.         xaProperties.put("user", readUser);
  37.         xaProperties.put("password", readPassword);
  38.         xaProperties.put("url", readJdbcUrl);
  39.         xaProperties.put("pinGlobalTxToPhysicalConnection", true);
  40.         dataSourceA.setXaProperties(xaProperties);
  41.         dataSourceA.setMaxPoolSize(readMaxPoolSize);
  42.         dataSourceA.setMinPoolSize(readMinPoolSize);
  43.         dataSourceA.setMaxIdleTime(readMaxIdleTime);
  44.         dataSourceA.setTestQuery("SELECT 1");
  45.         return dataSourceA;
  46.     }



点击(此处)折叠或打开

  1. @Configuration
  2. public class DynamicTransactionManagerElConfig {

  4.     // @Autowired
  5.     // @Qualifier("platformTomcat")
  6.     // private DataSource platformTomcat;
  7.     //
  8.     // @Autowired
  9.     // @Qualifier("platformReadTomcat")
  10.     // private DataSource platformReadTomcat;

  12.     @Autowired
  13.     @Qualifier("dataSourceA")
  14.     private DataSource dataSourceA;

  16.     @Autowired
  17.     @Qualifier("dataSourceB")
  18.     private DataSource dataSourceB;

  20.     @Bean(name = "dataSource")
  21.     public DynamicDataSource dataSource() {
  22.         DynamicDataSource dataSource = new DynamicDataSource();
  23.         Map<Object, Object> targetDataSources = new HashMap<>();
  24.         targetDataSources.put("master", dataSourceA);
  25.         targetDataSources.put("slave", dataSourceB);
  26.         dataSource.setTargetDataSources(targetDataSources);
  27.         dataSource.setDefaultTargetDataSource(dataSourceA);
  28.         return dataSource;
  29.     }

  31.     @Bean(name = "jdbcTemplate")
  32.     public JdbcTemplate jdbcTemplate(DynamicDataSource dataSource) {
  33.         JdbcTemplate jdbcTemplate = new JdbcTemplate();
  34.         jdbcTemplate.setDataSource(dataSource);
  35.         return jdbcTemplate;
  36.     }

  38.     @Bean(name = "jdbcReadTemplate")
  39.     public JdbcTemplate jdbcReadTemplate(DynamicDataSource dataSource) {
  40.         JdbcTemplate jdbcReadTemplate = new JdbcTemplate();
  41.         jdbcReadTemplate.setDataSource(dataSource);
  42.         return jdbcReadTemplate;
  43.     }

  45.     @Bean(name = "atomikosTransactionManager", initMethod = "init", destroyMethod = "close")
  46.     public UserTransactionManager atomikosTransactionManager() {
  47.         UserTransactionManager atomikosTransactionManager = new UserTransactionManager();
  48.         atomikosTransactionManager.setForceShutdown(true);
  49.         return atomikosTransactionManager;
  50.     }

  52.     @Bean(name = "atomikosUserTransaction")
  53.     public UserTransactionImp atomikosUserTransaction() {
  54.         UserTransactionImp atomikosUserTransaction = new UserTransactionImp();
  55.         try {
  56.             atomikosUserTransaction.setTransactionTimeout(300);
  57.         }
  58.         catch (SystemException e) {
  59.             e.printStackTrace();
  60.         }
  61.         return atomikosUserTransaction;
  62.     }

  64.     // @Bean(name = "transactionManager")
  65.     // public DataSourceTransactionManager transactionManager(DynamicDataSource
  66.     // dataSource) {
  67.     // DataSourceTransactionManager transactionManager = new
  68.     // DataSourceTransactionManager();
  69.     // transactionManager.setDataSource(dataSource);
  70.     // return transactionManager;
  71.     // }

  73.     @Bean(name = "transactionManager")
  74.     public JtaTransactionManager transactionManager(UserTransactionManager atomikosTransactionManager,
  75.             UserTransactionImp atomikosUserTransaction) {
  76.         JtaTransactionManager transactionManager = new JtaTransactionManager();
  77.         transactionManager.setTransactionManager(atomikosTransactionManager);
  78.         transactionManager.setUserTransaction(atomikosUserTransaction);
  79.         transactionManager.setAllowCustomIsolationLevels(true);
  80.         return transactionManager;
  81.     }

  83. }
三：AOP 注解方式数据源动态切换


点击(此处)折叠或打开

  1. @Retention(RetentionPolicy.RUNTIME)
  2. @Target(ElementType.METHOD)
  3. @Documented
  4. public @interface DataSource {
  5.     String value();
  6. }


点击(此处)折叠或打开

  1. public class DynamicDataSource extends AbstractRoutingDataSource {

  3.     @Override
  4.     protected Object determineCurrentLookupKey() {
  5.         return DataSourceContextHolder.getDataSource();
  6.     }

  8. }

点击(此处)折叠或打开

  1. public class DataSourceContextHolder {

  3.     private static final ThreadLocal<String> contextHolder = new ThreadLocal<String>();

  5.     public static void setDataSource(String dataSource) {
  6.         contextHolder.set(dataSource);
  7.     }

  9.     public static String getDataSource() {
  10.         return contextHolder.get();
  11.     }

  13.     public static void removeDataSource() {
  14.         contextHolder.remove();
  15.     }

  17. }

点击(此处)折叠或打开

  1. @Aspect
  2. @Order(1)
  3. @Component
  4. public class DataSourceAspect {

  6.     @Pointcut("@annotation(com.gemdale.ghome.business.async.deal.center.demo.datasource.DataSource)")
  7.     public void dataSourcePointCut() {
  8.     };

  10.     @Before("dataSourcePointCut()")
  11.     public void before(JoinPoint joinPoint) {

  13.         System.out.println("=============dataSourcePointCut:before=============");
  14.         Object target = joinPoint.getTarget();
  15.         String method = joinPoint.getSignature().getName();

  17.         // Class[] classz = target.getClass().getInterfaces();
  18.         Class<?> classz = target.getClass();
  19.         Class<?>[] parameterTypes = ((MethodSignature) joinPoint.getSignature()).getMethod().getParameterTypes();

  21.         try {
  22.             // Method m = classz[0].getMethod(method, parameterTypes);
  23.             Method m = classz.getMethod(method, parameterTypes);
  24.             if (null != m && m.isAnnotationPresent(DataSource.class)) {
  25.                 DataSource dataSource = m.getAnnotation(DataSource.class);
  26.                 DataSourceContextHolder.setDataSource(dataSource.value());

  28.                 System.out.println("=============dataSource:" + dataSource.value());
  29.             }
  30.         }
  31.         catch (Exception e) {
  32.             e.printStackTrace();
  33.         }
  34.     }

  36. }
四：数据源动态切换实例

点击(此处)折叠或打开

  1. @Repository("gmcSmsInfoDaoImpl")
  2. public class GmcSmsInfoDaoImpl extends BaseDaoSupport implements GmcSmsInfoDAO{

  4.     /* (non-Javadoc)
  5.      * @see com.gemdale.ghome.business.async.deal.center.demo.GmcSmsInfoDAO#addMaster(com.gemdale.ghome.business.async.deal.center.demo.GmcSmsInfo)
  6.      */
  7.     @Override
  8.     @DataSource("master")
  9.     public Integer addMaster(GmcSmsInfo smsInfo) throws FrameworkDAOException {
  10.         return save(smsInfo);
  11.     }

  13.     /* (non-Javadoc)
  14.      * @see com.gemdale.ghome.business.async.deal.center.demo.GmcSmsInfoDAO#addSlave(com.gemdale.ghome.business.async.deal.center.demo.GmcSmsInfo)
  15.      */
  16.     @Override
  17.     @DataSource("slave")
  18.     public Integer addSlave(GmcSmsInfo smsInfo) throws FrameworkDAOException {
  19.         return save(smsInfo);
  20.     }


  23. }


五：分布式事务下的数据源动态切换
情况一：事务下唯一数据源切换



点击(此处)折叠或打开

  1. @Transactional(rollbackFor={Exception.class,RuntimeException.class})
  2. @DataSource("master")
  3.     public GmcSmsInfo addMaster(GmcSmsInfo smsInfo) throws BusinessServiceException {
  4.         try {
  5.             smsInfo.setSmsId(gmcSmsInfoDaoImpl.save(smsInfo));
  6.         }
  7.         catch (FrameworkDAOException e) {
  8.             throw new BusinessServiceException(e);
  9.         }
  10.         return smsInfo;
  11.     }

@Transactional 通过数据库的connection来建立事务的，为了保证数据源能顺利切换，要保证@DataSource优先于@Transactional执行。   实现办法 在DataSourceAspect 切面上增加注解@Order(1).   此处需要了解切面的层次和执行顺序等相关知识。   @Transactionl,@DataSource 在service层的同一个方法上。

情况二：事务下多数据源切换

service类

点击(此处)折叠或打开

  1. @Autowired
  2.     private GmcSmsInfoDAO gmcSmsInfoDaoImpl;

  4.     @Autowired
  5.     @Qualifier("transactionManager")
  6.     private JtaTransactionManager transactionManager;


  9.     public void addMasterAndSlave(GmcSmsInfo smsInfo) throws BusinessServiceException {
  10.         UserTransaction userTransaction = transactionManager.getUserTransaction();
  11.         try {
  12.             userTransaction.begin();
  13.             gmcSmsInfoDaoImpl.addMaster(smsInfo);
  14.             smsInfo = new GmcSmsInfo();

  16.             smsInfo.setChannel("test2");
  17.             smsInfo.setContent("test2");
  18.             smsInfo.setStatus("001");
  19.             smsInfo.setCreateDate(Calendar.getInstance().getTime());
  20.             smsInfo.setMobile("88888888");
  21.             gmcSmsInfoDaoImpl.addSlave(smsInfo);
  22.             userTransaction.commit();
  23.         }
  24.         catch (Exception e) {
  25.             try {
  26.                 userTransaction.rollback();
  27.             }
  28.             catch (IllegalStateException e1) {
  29.                 e1.printStackTrace();
  30.             }
  31.             catch (SecurityException e1) {
  32.                 e1.printStackTrace();
  33.             }
  34.             catch (SystemException e1) {
  35.                 e1.printStackTrace();
  36.             }
  37.             throw new BusinessServiceException(e);
  38.         }

  40.     }

dao实现类


点击(此处)折叠或打开

  1. /* (non-Javadoc)
  2.      * @see com.gemdale.ghome.business.async.deal.center.demo.GmcSmsInfoDAO#addMaster(com.gemdale.ghome.business.async.deal.center.demo.GmcSmsInfo)
  3.      */
  4.     @Override
  5.     @DataSource("master")
  6.     public Integer addMaster(GmcSmsInfo smsInfo) throws FrameworkDAOException {
  7.         return save(smsInfo);
  8.     }

  10.     /* (non-Javadoc)
  11.      * @see com.gemdale.ghome.business.async.deal.center.demo.GmcSmsInfoDAO#addSlave(com.gemdale.ghome.business.async.deal.center.demo.GmcSmsInfo)
  12.      */
  13.     @Override
  14.     @DataSource("slave")
  15.     public Integer addSlave(GmcSmsInfo smsInfo) throws FrameworkDAOException {
  16.         return save(smsInfo);
  17.     }

注意，此处采用了编程式来实现事务，注解式暂时还没有好的解决方法，欢迎大家讨论分享。  此处的@DataSource 放在了dao的实现层。


来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/28624388/viewspace-2137095/ ，如需转载，请注明出处，否则将追究法律责任。

1
0
上一篇： Spring 多数据源 AOP 动态切换
下一篇： CentOS6.5 64bit + MariaDB 10.0.30 + Galera Cluster 10.0.30 集群部署安装
请登录后发表评论 登录
全部评论
百联达
专家姓名：耿冲
15年以上互联网经验，先后从事过制造业，证券业，物业行业和物流行业信息系统和互联网产品的设计落地实施，8年系统架构经验。最近关注企业数字化转型、财务数字化服务共享、大语言模型、Kubernetes微服务架构和Istio微服务治理框架。

注册时间：2013-02-05

  • 博文量
    332
  • 访问量
    1265642

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593