zoukankan      html  css  js  c++  java
  • Atomikos 中文说明文档【转】

    Atomikos 翻译文档(英文文档来源:下载安装包中START_HERE.html)

                                     ----译者:周枫

    请尊重劳动成果,转载请标明,英语水平有限,如有不准确地方请在评论中指出,谢谢

    官网地址:http://www.atomikos.com/Main/WebHome

    使用版本:AtomikosTransactionsEssentials-3.7.2

    感谢您使用Atomikos,下面的说明文档可以让您正确使用,如果您有任何问题或者反馈,请访问我们的帮助网页http://www.atomikos.com/Main/SupportOverview,或者给我们发送邮件sales@atomikos.com

    什么是Atomikos TransactionsEssentials

             Atomikos TransactionsEssentials 是一个为Java平台提供增值服务的并且开源类事务管理器,以下是包括在这个开源版本中的一些功能:

    l  全面崩溃 / 重启恢复

    l  兼容标准的SUN公司JTA API

    l  嵌套事务

    l  为XA和非XA提供内置的JDBC适配器

    注释:XA:XA协议由Tuxedo首先提出的,并交给X/Open组织,作为资源管理器(数据库)与事务管理器的接口标准。目前,Oracle、Informix、DB2和Sybase等各大数据库厂家都提供对XA的支持。XA协议采用两阶段提交方式来管理分布式事务。XA接口提供资源管理器与事务管理器之间进行通信的标准接口。XA协议包括两套函数,以xa_开头的及以ax_开头的。

    以下的函数使事务管理器可以对资源管理器进行的操作:

      1)xa_open,xa_close:建立和关闭与资源管理器的连接。

      2)xa_start,xa_end:开始和结束一个本地事务。

      3)xa_prepare,xa_commit,xa_rollback:预提交、提交和回滚一个本地事务。

      4)xa_recover:回滚一个已进行预提交的事务。

      5)ax_开头的函数使资源管理器可以动态地在事务管理器中进行注册,并可以对XID(TRANSACTION IDS)进行操作。

      6)ax_reg,ax_unreg;允许一个资源管理器在一个TMS(TRANSACTION MANAGER SERVER)中动态注册或撤消注册。

    l  内置的JMS适配器XA-capable JMS队列连接器

    注释:JMS:jms即Java消息服务(Java Message Service)应用程序接口是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。Java消息服务是一个与具体平台无关的API,绝大多数MOM提供商都对JMS提供支持。

    l  通过XA API兼容第三方适配器

    l  更好的整合您的项目

    l  集成Hibernate

    如何使用Atomikos TransactionsEssentials

             Atomikos TransactionsEssentials 是一个可靠的库,可以加入到您的Java应用程序,也就是说为了使用这个产品,您必须添加一些jar文件(包括在dist和lib文件夹下)到您的应用程序或者应用程序服务器。

             请注意:Atomikos TransactionsEssentials是一个非常快速的嵌入式事务管理器,这就意味着,您不需要另外启动一个单独的事务管理器进程(不要查找任何的bin文件夹)。相反,您的应用服务器将有它自己的intra-VM事务管理器。

             配置需求:至少Java1.5 jdk,并且最少128M的内存

             性能优化:尽管这个软件有着很大的优势,但是想要更好的发挥其作用,可以按以下的方法优化:

    l  更高的内存,意味着更高的吞吐量(每秒的事务数目)

    l  使连接池尽可能的大

    l  一旦你不需要的连接请马上关闭它们。不要把你的应用程序放在缓存里,让内部连接池为你做这些,这将促使更高效的连接使用

    l  不要让活动的事务闲置:终止所有情况下的事务,尤其是在异常报错情况下的事务。这将减少数据库的锁定时间,并且最大效率的处理启用的使用。

    如果想获取这些细节的更多信息,也要参阅文档说明部分。

    值得注意的是,在我们所有的压力测试中,Atomikos TransactionsEssentials比J2EE的web容器更高效的吞吐量。这些测量值包括日志记录的高效的事务状态,同样,在我们所有的测量中,包括XA和non-XA,高效的效率是一样的。

             在J2SE中使用Atomikos Transactions Essentials,只需要按以下步骤

    1. 将idst和lib中的jar包全部放入的项目中
    2. 创建或者自定义你应用的transactions.properties(或者jta.properties)文件(事务管理器的配置),然后将它放入到classpath中,安装文件夹中包涵一个实例文件;在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 implementation class of the transaction core.
      # NOTE: there is no default for this, so it MUST be specified! 
      # 
      com.atomikos.icatch.service=com.atomikos.icatch.standalone.UserTransactionServiceFactory
      
              
      # Set base name of file where messages are output 
      # (also known as the 'console file').
      #
      # 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 default timeout (in milliseconds) for local transactions
      #
      # com.atomikos.icatch.default_jta_timeout = 10000
      
      # 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.
      #
      # 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?
      # Set this to false if you want commits to be ordered according
      # to the order in which resources are added to the transaction.
      #
      # NOTE: threads are reused on JDK 1.5 or higher. 
      # For JDK 1.4, thread reuse is enabled as soon as the 
      # concurrent backport is in the classpath - see 
      # http://mirrors.ibiblio.org/pub/mirrors/maven2/backport-util-concurrent/backport-util-concurrent/
      #
      # com.atomikos.icatch.threaded_2pc=false
      
      # Should shutdown of the VM trigger shutdown of the transaction core too?
      #
      # com.atomikos.icatch.force_shutdown_on_vm_exit=false
      复制代码
    3. 在你的应用程序中,创建一个实例com.atomikos.icatch.jta.UserTransactionImp或者com.atomikos.icatch.jta.UserTransactionManager(使用默认的无参数构造函数)
      复制代码
      /**
       * Copyright (C) 2000-2010 Atomikos <info@atomikos.com>
       *
       * This code ("Atomikos TransactionsEssentials"), by itself,
       * is being distributed under the
       * Apache License, Version 2.0 ("License"), a copy of which may be found at
       * http://www.atomikos.com/licenses/apache-license-2.0.txt .
       * You may not use this file except in compliance with the License.
       *
       * While the License grants certain patent license rights,
       * those patent license rights only extend to the use of
       * Atomikos TransactionsEssentials by itself.
       *
       * This code (Atomikos TransactionsEssentials) contains certain interfaces
       * in package (namespace) com.atomikos.icatch
       * (including com.atomikos.icatch.Participant) which, if implemented, may
       * infringe one or more patents held by Atomikos.
       * It should be appreciated that you may NOT implement such interfaces;
       * licensing to implement these interfaces must be obtained separately from Atomikos.
       *
       * Unless required by applicable law or agreed to in writing, software
       * distributed under the License is distributed on an "AS IS" BASIS,
       * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
       */
      
      package com.atomikos.icatch.jta;
      
      import java.io.Serializable;
      
      import javax.naming.NamingException;
      import javax.naming.Reference;
      import javax.naming.Referenceable;
      import javax.transaction.NotSupportedException;
      import javax.transaction.SystemException;
      import javax.transaction.TransactionManager;
      import javax.transaction.UserTransaction;
      
      import com.atomikos.icatch.admin.imp.SimpleLogAdministrator;
      import com.atomikos.icatch.config.TSInitInfo;
      import com.atomikos.icatch.config.UserTransactionService;
      import com.atomikos.icatch.config.UserTransactionServiceImp;
      import com.atomikos.util.SerializableObjectFactory;
      
      /**
       * 
       * 
       * Our UserTransaction implementation for J2SE transactions. This class is
       * special in that it automatically starts up and recover the transaction
       * service on first use. <b>Note: don't use this class in J2EE applications in
       * order to avoid starting different transaction engines in the same application
       * server! J2EE applications should use J2eeUserTransaction instead.</b>
       */
      
      public class UserTransactionImp implements UserTransaction, Serializable,
              Referenceable
      {
          private transient TransactionManager txmgr_;
      
          /**
           * No-argument constructor.
           */
      
          public UserTransactionImp ()
          {
          }
      
          /**
           * Referenceable mechanism requires later setup of txmgr_, otherwise binding
           * into JNDI already requires that TM is running.
           */
      
          private void checkSetup ()
          {
      
              // REMOVED FOLLOWING IF CHECK: DON'T CACHE THE TXMGR TO MAKE INSTANCES
              // RESILIENT TO RESTART IN TOMCAT. OTHERWISE, CLIENT APPS SEE THEIR
              // USERTX REFERENCES INVALIDATED AND THIS IS INTOLERABLE
              // if ( txmgr_ == null ) {
              // txmgr_ = TransactionManagerImp.getTransactionManager();
      
              synchronized ( TransactionManagerImp.class ) {
      
                  txmgr_ = TransactionManagerImp.getTransactionManager ();
      
                  // FOLLOWING COMMENTED OUT: NEW RECOVERY IN 2.0 ALLOWS US TO START
                  // THE TM
                  // IF NOT ALREADY RUNNING!!!
                  // if ( txmgr_ == null )
                  // throw new RuntimeException ( "No transaction monitor installed?"
                  // );
      
                  // NEW FROM 2.0: if TM is not running, just start it. Any resources
                  // can be registered later.
                  if ( txmgr_ == null ) {
                      UserTransactionService uts = new UserTransactionServiceImp ();
                      TSInitInfo info = uts.createTSInitInfo ();
                      uts.registerLogAdministrator ( SimpleLogAdministrator
                              .getInstance () );
                      uts.init ( info );
                      txmgr_ = TransactionManagerImp.getTransactionManager ();
                  }
      
              }
      
              // }
          }
      
          /**
           * @see javax.transaction.UserTransaction
           */
      
          public void begin () throws NotSupportedException, SystemException
          {
              checkSetup ();
              txmgr_.begin ();
          }
      
          /**
           * @see javax.transaction.UserTransaction
           */
      
          public void commit () throws javax.transaction.RollbackException,
                  javax.transaction.HeuristicMixedException,
                  javax.transaction.HeuristicRollbackException,
                  javax.transaction.SystemException, java.lang.IllegalStateException,
                  java.lang.SecurityException
          {
              checkSetup ();
              txmgr_.commit ();
          }
      
          /**
           * @see javax.transaction.UserTransaction
           */
      
          public void rollback () throws IllegalStateException, SystemException,
                  SecurityException
          {
              checkSetup ();
              txmgr_.rollback ();
          }
      
          /**
           * @see javax.transaction.UserTransaction
           */
      
          public void setRollbackOnly () throws IllegalStateException,
                  SystemException
          {
              checkSetup ();
              txmgr_.setRollbackOnly ();
          }
      
          /**
           * @see javax.transaction.UserTransaction
           */
      
          public int getStatus () throws SystemException
          {
              checkSetup ();
              return txmgr_.getStatus ();
          }
      
          /**
           * @see javax.transaction.UserTransaction
           */
      
          public void setTransactionTimeout ( int seconds ) throws SystemException
          {
              checkSetup ();
              txmgr_.setTransactionTimeout ( seconds );
          }
      
          //
          //
          // IMPLEMENTATION OF REFERENCEABLE
          //
          //
      
          public Reference getReference () throws NamingException
          {
              return SerializableObjectFactory.createReference ( this );
          }
      }
      复制代码
      复制代码
      /**
       * Copyright (C) 2000-2010 Atomikos <info@atomikos.com>
       *
       * This code ("Atomikos TransactionsEssentials"), by itself,
       * is being distributed under the
       * Apache License, Version 2.0 ("License"), a copy of which may be found at
       * http://www.atomikos.com/licenses/apache-license-2.0.txt .
       * You may not use this file except in compliance with the License.
       *
       * While the License grants certain patent license rights,
       * those patent license rights only extend to the use of
       * Atomikos TransactionsEssentials by itself.
       *
       * This code (Atomikos TransactionsEssentials) contains certain interfaces
       * in package (namespace) com.atomikos.icatch
       * (including com.atomikos.icatch.Participant) which, if implemented, may
       * infringe one or more patents held by Atomikos.
       * It should be appreciated that you may NOT implement such interfaces;
       * licensing to implement these interfaces must be obtained separately from Atomikos.
       *
       * Unless required by applicable law or agreed to in writing, software
       * distributed under the License is distributed on an "AS IS" BASIS,
       * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
       */
      
      package com.atomikos.icatch.jta;
      
      import java.io.Serializable;
      
      import javax.naming.NamingException;
      import javax.naming.Reference;
      import javax.naming.Referenceable;
      import javax.transaction.HeuristicMixedException;
      import javax.transaction.HeuristicRollbackException;
      import javax.transaction.InvalidTransactionException;
      import javax.transaction.NotSupportedException;
      import javax.transaction.RollbackException;
      import javax.transaction.SystemException;
      import javax.transaction.Transaction;
      import javax.transaction.TransactionManager;
      import javax.transaction.UserTransaction;
      
      import com.atomikos.icatch.config.TSInitInfo;
      import com.atomikos.icatch.config.UserTransactionService;
      import com.atomikos.icatch.config.UserTransactionServiceImp;
      import com.atomikos.util.SerializableObjectFactory;
      
      /**
       * 
       * 
       * 
       * 
       * 
       * A straightforward, zero-setup implementation of a transaction manager. J2SE
       * applications can use an instance of this class to get a handle to the
       * transaction manager, and automatically startup or recover the transaction
       * service on first use. <b>J2EE applications should NOT use this class in order
       * to avoid the concurrent use of different transaction services. For J2EE
       * applications, we have the class J2eeTransactionManager instead.</b>
       */
      public class UserTransactionManager implements TransactionManager,
              Serializable, Referenceable, UserTransaction
      {
          private static final long serialVersionUID = -655789038710288096L;
      
          private transient TransactionManagerImp tm;
          
          private UserTransactionService uts;
      
          private boolean forceShutdown;
          
          private boolean startupTransactionService;
          
          private boolean closed;
      
          private void checkSetup () throws SystemException
          {
              if ( closed ) throw new SystemException ( "This UserTransactionManager instance was closed already. Call init() to reuse if desired." );
                  
              synchronized ( TransactionManagerImp.class ) {
      
                  tm = (TransactionManagerImp) TransactionManagerImp
                          .getTransactionManager ();
                  if ( tm == null ) {
                      // not initialized -> startup TM
                      // System.out.println ( "STARTING UP TM!!!!!!");
                         if ( getStartupTransactionService() ) {
                          uts = new UserTransactionServiceImp ();
                          TSInitInfo info = uts.createTSInitInfo ();
                          uts.init ( info );
                          tm = (TransactionManagerImp) TransactionManagerImp
                                  .getTransactionManager ();
                         }
                         else {
                             throw new SystemException ( "Transaction service not running" );
                         }
                  }
              }
          }
          
          public UserTransactionManager()
          {
                  //startup by default, to have backward compatibility
                  this.startupTransactionService = true;
                  this.closed = false;
          }
          
          /**
           * Sets whether the transaction service should be 
           * started if not already running. 
           * @param startup
           */
          public void setStartupTransactionService ( boolean startup )
          {
                  this.startupTransactionService = startup;
          }
          
          /**
           * Returns true if the transaction service will 
           * be started if not already running.
           * @return
           */
          public boolean getStartupTransactionService()
          {
                  return this.startupTransactionService;
          }
          
          /**
           * Performs initialization if necessary.
           * This will startup the TM (if not running)
           * and perform recovery, unless <b>getStartupTransactionService</b>
           * returns false.
           * 
           * @throws SystemException
           */
          
          public void init() throws SystemException
          {
                  closed = false;
                  checkSetup();
          }
      
          /**
           * @see javax.transaction.TransactionManager#begin()
           */
          public void begin () throws NotSupportedException, SystemException
          {
              checkSetup ();
              tm.begin ();
      
          }
          
          public boolean getForceShutdown()
          {
                  return forceShutdown;
          }
          
          /**
           * Sets the force shutdown mode to use during close.
           * @param value 
           */
          public void setForceShutdown ( boolean value )
          {
                  this.forceShutdown = value;
          }
      
          /**
           * @see javax.transaction.TransactionManager#commit()
           */
          public void commit () throws RollbackException, HeuristicMixedException,
                  HeuristicRollbackException, SecurityException,
                  IllegalStateException, SystemException
          {
              checkSetup ();
              tm.commit ();
      
          }
      
          /**
           * @see javax.transaction.TransactionManager#getStatus()
           */
          public int getStatus () throws SystemException
          {
              checkSetup ();
              return tm.getStatus ();
          }
      
          /**
           * @see javax.transaction.TransactionManager#getTransaction()
           */
          public Transaction getTransaction () throws SystemException
          {
              checkSetup ();
              return tm.getTransaction ();
          }
      
          /**
           * @see javax.transaction.TransactionManager#resume(javax.transaction.Transaction)
           */
          public void resume ( Transaction tx ) throws InvalidTransactionException,
                  IllegalStateException, SystemException
          {
              checkSetup ();
              tm.resume ( tx );
      
          }
      
          /**
           * @see javax.transaction.TransactionManager#rollback()
           */
          public void rollback () throws IllegalStateException, SecurityException,
                  SystemException
          {
              checkSetup ();
              tm.rollback ();
      
          }
      
          /**
           * @see javax.transaction.TransactionManager#setRollbackOnly()
           */
          public void setRollbackOnly () throws IllegalStateException,
                  SystemException
          {
              checkSetup ();
              tm.setRollbackOnly ();
      
          }
      
          /**
           * @see javax.transaction.TransactionManager#setTransactionTimeout(int)
           */
          public void setTransactionTimeout ( int secs ) throws SystemException
          {
              checkSetup ();
              tm.setTransactionTimeout ( secs );
      
          }
      
          /**
           * @see javax.transaction.TransactionManager#suspend()
           */
          public Transaction suspend () throws SystemException
          {
              checkSetup ();
              return tm.suspend ();
          }
      
          /**
           * @see javax.naming.Referenceable#getReference()
           */
          public Reference getReference () throws NamingException
          {
              return SerializableObjectFactory.createReference ( this );
          }
          
          /**
           * Closes the transaction service, but only if it was 
           * implicitly started via this instance.
           * In other words, if the transaction service was started
           * in another way then this method will not do anything.
           *
           */
          public void close()
          {
                  if ( uts != null ) {
                      uts.shutdown ( forceShutdown );
                      uts = null;
                  }
                  closed = true;
          }
      
      }
      复制代码
    4. 对于JDBC,使用我们的一个实例com.atomikos.jdbc.AtomikosDataSourceBean或者,对于non-XA驱动,可以使用com.atomikos.jdbc.nonxa.AtomikosNonXADataSourceBean
      复制代码
      /**
       * Copyright (C) 2000-2010 Atomikos <info@atomikos.com>
       *
       * This code ("Atomikos TransactionsEssentials"), by itself,
       * is being distributed under the
       * Apache License, Version 2.0 ("License"), a copy of which may be found at
       * http://www.atomikos.com/licenses/apache-license-2.0.txt .
       * You may not use this file except in compliance with the License.
       *
       * While the License grants certain patent license rights,
       * those patent license rights only extend to the use of
       * Atomikos TransactionsEssentials by itself.
       *
       * This code (Atomikos TransactionsEssentials) contains certain interfaces
       * in package (namespace) com.atomikos.icatch
       * (including com.atomikos.icatch.Participant) which, if implemented, may
       * infringe one or more patents held by Atomikos.
       * It should be appreciated that you may NOT implement such interfaces;
       * licensing to implement these interfaces must be obtained separately from Atomikos.
       *
       * Unless required by applicable law or agreed to in writing, software
       * distributed under the License is distributed on an "AS IS" BASIS,
       * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
       */
      
      package com.atomikos.jdbc;
      import java.util.Enumeration;
      import java.util.Properties;
      
      import javax.sql.XADataSource;
      
      import com.atomikos.beans.PropertyUtils;
      import com.atomikos.datasource.RecoverableResource;
      import com.atomikos.datasource.xa.jdbc.JdbcTransactionalResource;
      import com.atomikos.icatch.system.Configuration;
      import com.atomikos.jdbc.AbstractDataSourceBean;
      import com.atomikos.util.ClassLoadingHelper;
      
       /**
       * The preferred class for using Atomikos connection pooling. Use an instance of
       * this class if you want to use Atomikos JTA-enabled connection pooling. All
       * you need to do is construct an instance and set the required properties as
       * outlined below. The resulting bean will automatically register with the
       * transaction service (for recovery) and take part in active transactions.
       * All SQL done over connections (gotten from this class) will participate in JTA transactions.
       */
      
      public class AtomikosDataSourceBean 
      extends AbstractDataSourceBean
      {
          
          
          private static final long serialVersionUID = 1L;
          
          private Properties xaProperties = null;
          private String xaDataSourceClassName;
          private transient XADataSource xaDataSource;
          public AtomikosDataSourceBean() 
          {
              this.xaProperties = new Properties();
          }
          
          protected String printXaProperties()
          {
              StringBuffer ret = new StringBuffer();
              if ( xaProperties != null ) {
                  Enumeration it = xaProperties.propertyNames();
                  ret.append ( "[" );
                  boolean first = true;
                  while ( it.hasMoreElements() ) {
                      if ( ! first ) ret.append ( "," );
                      String name = ( String ) it.nextElement();
                      String value = xaProperties.getProperty( name);
                      ret.append ( name ); ret.append ( "=" ); ret.append ( value );
                      first = false;
                  }
                  ret.append ( "]" );
              }
              return ret.toString();
          }
          
          /**
           * Gets the properties used to
           * configure the XADataSource.  
           */
          
          public Properties getXaProperties()
          {
              return xaProperties;
          }
      
          /**
           * Sets the properties (name,value pairs) used to
           * configure the XADataSource. Required, unless you call setXaDataSource directly.
           * 
           * @param xaProperties 
           * 
           *
           */
          public void setXaProperties ( Properties xaProperties ) 
          {
              this.xaProperties = xaProperties;
          }
      
          /**
           * Get the XADataSource class name.
           */
          public String getXaDataSourceClassName() 
          {
              return xaDataSourceClassName;
          }
      
          /**
           * Sets the fully qualified underlying XADataSource class name. Required, unless you 
           * call setXaDataSource directly.
           * 
           * @param xaDataSourceClassName
           */
          public void setXaDataSourceClassName ( String xaDataSourceClassName ) 
          {
              this.xaDataSourceClassName = xaDataSourceClassName;
          }
          
          /**
           * Gets the configured XADataSource (if any).
           * @return The instance, or null if none.
           */
          
          public XADataSource getXaDataSource()
          {
              return xaDataSource;
          }
          
          /**
           * Sets the XADataSource directly - instead of providing the xaDataSourceClassName and xaProperties.
           * @param xaDataSource
           */
          public void setXaDataSource(XADataSource xaDataSource)
          {
              this.xaDataSource = xaDataSource;
          }
          
          
          protected com.atomikos.datasource.pool.ConnectionFactory doInit() throws Exception 
          {
              if (xaDataSource == null)
              {
                  if (xaDataSourceClassName == null)
                      throwAtomikosSQLException("Property 'xaDataSourceClassName' cannot be null");
                  if (xaProperties == null)
                      throwAtomikosSQLException("Property 'xaProperties' cannot be null");
              }
              
              
              if ( Configuration.isInfoLoggingEnabled() ) Configuration.logInfo(
                      this + ": initializing with [" +
                      " xaDataSourceClassName=" + xaDataSourceClassName + "," +
                      " uniqueResourceName=" + getUniqueResourceName() + "," +
                      " maxPoolSize=" + getMaxPoolSize() + "," +
                      " minPoolSize=" + getMinPoolSize() + "," +
                      " borrowConnectionTimeout=" + getBorrowConnectionTimeout() + "," +
                      " maxIdleTime=" + getMaxIdleTime() + "," +
                      " reapTimeout=" + getReapTimeout() + "," +
                      " maintenanceInterval=" + getMaintenanceInterval() + "," +
                      " testQuery=" + getTestQuery() + "," +
                      " xaProperties=" + printXaProperties() + 
                      " loginTimeout=" + getLoginTimeout() +
                      "]"
                      );
              
              
                  if (xaDataSource == null)
                  {
                      Class xadsClass = null;
                      try {
                          xadsClass = ClassLoadingHelper.loadClass ( getXaDataSourceClassName() );
                      } catch ( ClassNotFoundException nf ) {
                          AtomikosSQLException.throwAtomikosSQLException ( "The class '" + getXaDataSourceClassName() +
                                  "' specified by property 'xaDataSourceClassName' could not be found in the classpath. Please make sure the spelling is correct, and that the required jar(s) are in the classpath." , nf );
                           
                      }
                      Object driver =  xadsClass.newInstance();
                      if ( ! ( driver instanceof XADataSource ) ) {
                          AtomikosSQLException.throwAtomikosSQLException (
                                   "The class '" + getXaDataSourceClassName() +
                                      "' specified by property 'xaDataSourceClassName' does not implement the required interface javax.jdbc.XADataSource. Please make sure the spelling is correct, and check your JDBC driver vendor's documentation." 
                          );
                      }
                      xaDataSource = (XADataSource) driver;
                      xaDataSource.setLoginTimeout ( getLoginTimeout() );
                      xaDataSource.setLogWriter ( getLogWriter() );
                      PropertyUtils.setProperties(xaDataSource, xaProperties );
                  }
                  
                  JdbcTransactionalResource tr = new JdbcTransactionalResource(getUniqueResourceName() , xaDataSource);
                  com.atomikos.datasource.pool.ConnectionFactory cf = new com.atomikos.jdbc.AtomikosXAConnectionFactory(xaDataSource, tr, this);
                  Configuration.addResource ( tr );
                  
                  return cf;
          }
          
          protected void doClose() 
          {
              RecoverableResource res = Configuration.getResource ( getUniqueResourceName() );
              if ( res != null ) {
                  Configuration.removeResource ( getUniqueResourceName() );
                  //fix for case 26005
                  res.close();
              }
          }    
      
          public String toString() 
          {
              String ret = "AtomikosDataSoureBean";
              String name = getUniqueResourceName();
              if ( name != null ) {
                  ret = ret + " '" + name + "'";
              }
              return ret;
          }
      
      }
      复制代码
      复制代码
      /**
       * Copyright (C) 2000-2010 Atomikos <info@atomikos.com>
       *
       * This code ("Atomikos TransactionsEssentials"), by itself,
       * is being distributed under the
       * Apache License, Version 2.0 ("License"), a copy of which may be found at
       * http://www.atomikos.com/licenses/apache-license-2.0.txt .
       * You may not use this file except in compliance with the License.
       *
       * While the License grants certain patent license rights,
       * those patent license rights only extend to the use of
       * Atomikos TransactionsEssentials by itself.
       *
       * This code (Atomikos TransactionsEssentials) contains certain interfaces
       * in package (namespace) com.atomikos.icatch
       * (including com.atomikos.icatch.Participant) which, if implemented, may
       * infringe one or more patents held by Atomikos.
       * It should be appreciated that you may NOT implement such interfaces;
       * licensing to implement these interfaces must be obtained separately from Atomikos.
       *
       * Unless required by applicable law or agreed to in writing, software
       * distributed under the License is distributed on an "AS IS" BASIS,
       * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
       */
      
      package com.atomikos.jdbc.nonxa;
      
      
      import java.sql.Connection;
      import java.sql.SQLException;
      
      import com.atomikos.datasource.pool.ConnectionFactory;
      import com.atomikos.icatch.HeuristicMessage;
      import com.atomikos.icatch.system.Configuration;
      import com.atomikos.jdbc.AbstractDataSourceBean;
      import com.atomikos.util.DynamicProxy;
      
       /**
        * 
        * A Bean class for DataSource access to non-XA JDBC implementations. 
        * Instances are JTA transaction-aware and can rollback the work done 
        * over multiple connections (provided that all work was done in one and the same thread).
        * 
        *
        */
      public class AtomikosNonXADataSourceBean extends AbstractDataSourceBean 
      {
          private static final long serialVersionUID = 1L;
          
          private String url;
          
          private String user;
          
          private String password;
          
          private String driverClassName;
          
          private boolean readOnly;
          
          /**
           * Sets the URL to use for getting connections. Required.
           * 
           * @param url
           */
          
          public void setUrl ( String url ) 
          {
              this.url = url;
          }
          
          /**
           * Gets the URL to connect. 
           */
          
          public String getUrl()
          {
              return url;
          }
          
      
          /**
           * Marks this datasource as being used for read-only work. Optional.
           * 
           * Setting this to true will avoid warnings/errors upon recovery. ReadOnly mode
           * is intended to avoid XA configuration of databases where no updates are
           * being done.
           * 
           * @param readOnly Defaults to false.
           */
          
          public void setReadOnly ( boolean readOnly ) 
          {
              this.readOnly = readOnly;
          }
      
          /**
           * @return Whether or not this datasource is marked as readOnly.
           */
          
          public boolean getReadOnly()
          {
              return readOnly;
          }
      
          /**
           * @return The password.
           */
          
          public String getPassword ()
          {
              return password;
          }
      
          /**
           * Sets the password to use.
           * 
           * @param string
           */
          
          public void setPassword ( String string )
          {
              password = string;
          }
      
          /**
           * Set the user name to get connections with.
           * 
           * @param string
           */
          
          public void setUser ( String string )
          {
              user = string;
          }
      
          /**
           * @return The URL to connect with.
           */
          
          public String getUser ()
          {
              return user;
          }
          
          /**
           * 
           * @return The DriverManager class name.
           */
      
          public String getDriverClassName ()
          {
              return driverClassName;
          }
      
          /**
           * Sets the driver class name to be used by the DriverManager. Required.
           * 
           * @param string
           */
          public void setDriverClassName ( String string )
          {
              driverClassName = string;
          }
          
      
          protected void doClose() 
          {
              //nothing to do
          }
      
          protected ConnectionFactory doInit() throws Exception 
          {
              AtomikosNonXAConnectionFactory ret = null;
              if ( Configuration.isInfoLoggingEnabled() ) Configuration.logInfo(
                      this + ": initializing with [" +
                      " uniqueResourceName=" + getUniqueResourceName() + "," +
                      " maxPoolSize=" + getMaxPoolSize() + "," +
                      " minPoolSize=" + getMinPoolSize() + "," +
                      " borrowConnectionTimeout=" + getBorrowConnectionTimeout() + "," +
                      " maxIdleTime=" + getMaxIdleTime() + "," +
                      " reapTimeout=" + getReapTimeout() + "," +
                      " maintenanceInterval=" + getMaintenanceInterval() + "," +
                      " testQuery=" + getTestQuery() + "," +
                      " driverClassName=" + getDriverClassName() + "," +
                      " user=" + getUser() + "," +
                      " url=" + getUrl() + 
                      " loginTimeout=" + getLoginTimeout() +
                      "]"
                      );
              
              
              ret = new com.atomikos.jdbc.nonxa.AtomikosNonXAConnectionFactory ( this , url , driverClassName , user , password , getLoginTimeout() , readOnly ) ;
              ret.init();
              return ret;
          }
      
          public synchronized Connection getConnection ( HeuristicMessage hmsg ) throws SQLException
          {
              if ( Configuration.isInfoLoggingEnabled() ) Configuration.logInfo ( this + ": getConnection ( " + hmsg + " )..." );
              
              init();
              
              
              //let pool take care of reusing an existing handle
              Connection proxy = super.getConnection ( hmsg );          
      
              // here we are certain that proxy is not null -> increase the use count
              DynamicProxy dproxy = ( DynamicProxy ) proxy;
              com.atomikos.jdbc.nonxa.AtomikosThreadLocalConnection previous = (AtomikosThreadLocalConnection) dproxy.getInvocationHandler();
      
              previous.incUseCount();
              previous.addHeuristicMessage ( hmsg );
              if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( this + ": returning " + proxy );
              return proxy;
          }
      
          
      
          public String toString() 
          {
              String ret = "AtomikosNonXADataSourceBean";
              String name = getUniqueResourceName();
              if ( name != null ) {
                  ret = ret + " '" + name + "'";
              }
              return ret;
          }
          
      }
      复制代码
    5. 对于JMS,可以使用我们的实例com.atomikos.jms.AtomikosConnectionFactoryBean,com.atomikos.jms.extra.AbstractJmsSenderTemplate(发送信息时使用)com.atomikos.jms.extra.MessageDrivenContainer(接收时使用)
      复制代码
      /**
       * Copyright (C) 2000-2010 Atomikos <info@atomikos.com>
       *
       * This code ("Atomikos TransactionsEssentials"), by itself,
       * is being distributed under the
       * Apache License, Version 2.0 ("License"), a copy of which may be found at
       * http://www.atomikos.com/licenses/apache-license-2.0.txt .
       * You may not use this file except in compliance with the License.
       *
       * While the License grants certain patent license rights,
       * those patent license rights only extend to the use of
       * Atomikos TransactionsEssentials by itself.
       *
       * This code (Atomikos TransactionsEssentials) contains certain interfaces
       * in package (namespace) com.atomikos.icatch
       * (including com.atomikos.icatch.Participant) which, if implemented, may
       * infringe one or more patents held by Atomikos.
       * It should be appreciated that you may NOT implement such interfaces;
       * licensing to implement these interfaces must be obtained separately from Atomikos.
       *
       * Unless required by applicable law or agreed to in writing, software
       * distributed under the License is distributed on an "AS IS" BASIS,
       * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
       */
      
      package com.atomikos.jms;
      
      import java.io.Serializable;
      
      import javax.jms.ConnectionFactory;
      import javax.jms.JMSException;
      import javax.naming.NamingException;
      import javax.naming.Reference;
      import javax.naming.Referenceable;
      
      import com.atomikos.util.SerializableObjectFactory;
      
      /**
       * 
       * 
       * Common logic for the connection factory beans.
       *
       */
      
      public abstract class AbstractConnectionFactoryBean 
      implements Serializable, Referenceable, ConnectionFactory
      {
      
          protected String resourceName_;
          protected String xaFactoryJndiName_;
      
          protected AbstractConnectionFactoryBean ( )
          {
              this.xaFactoryJndiName_ = "";
              this.resourceName_ = "someUniqueName";
          }
          
          /**
           * Sets the JNDI name of the underlying XAConnectionFactory (optional). This is
           * optional and an alternative to directly supplying the required factory
           * through setXaConnectionFactory().
           * 
           * @param name
           *            The JNDI name where the XAConnectionFactory can be found.
           *            It is up to the client to make sure that the name exists and
       *            points to an existing XAConnectionFactory.
           */
          public void setXaFactoryJndiName ( String name ) 
          {
              xaFactoryJndiName_ = name;
          
          }
      
          /**
           * Retrieve the JNDI name where the XAConnectionFactory is expected.
           * 
           * @return String the name or an empty String if not set.
           */
          public String getXaFactoryJndiName() 
          {
              return xaFactoryJndiName_;
          }
      
          /**
           * Set the unique resource name for this factory (required). A unique
           * resource name is needed by the transaction service in order to register
           * and recover the underlying XA transactions. 
           * Note: the value you set here should not exceed 45 bytes in length.
           * 
           * <p><b>MQSeries NOTE:</b> For
           * IBM MQSeries, the name should include MQSeries_XA_RMI or the XA routines
           * will not work properly! 
           * 
           * @param name
           *            The unique resource name.
           */
          public void setResourceName ( String name ) 
          {
              resourceName_ = name;
          }
      
          /**
           * Get the resource name.
           * 
           * @return String the unique resource name as previously set.
           */
          public String getResourceName() 
          {
              return resourceName_;
          }
      
          public Reference getReference() throws NamingException 
          {
              return SerializableObjectFactory.createReference ( this );
          }
          
          /**
           * Initialization method to register the underlying resource for recovery
           * and other init code. 
           * 
           * @throws JMSException
           */
          
          public void init() throws JMSException
          {
                  checkSetup();
          }
          
          protected abstract void checkSetup() throws JMSException;
      
      }
      复制代码
      复制代码
      /**
       * Copyright (C) 2000-2010 Atomikos <info@atomikos.com>
       *
       * This code ("Atomikos TransactionsEssentials"), by itself,
       * is being distributed under the
       * Apache License, Version 2.0 ("License"), a copy of which may be found at
       * http://www.atomikos.com/licenses/apache-license-2.0.txt .
       * You may not use this file except in compliance with the License.
       *
       * While the License grants certain patent license rights,
       * those patent license rights only extend to the use of
       * Atomikos TransactionsEssentials by itself.
       *
       * This code (Atomikos TransactionsEssentials) contains certain interfaces
       * in package (namespace) com.atomikos.icatch
       * (including com.atomikos.icatch.Participant) which, if implemented, may
       * infringe one or more patents held by Atomikos.
       * It should be appreciated that you may NOT implement such interfaces;
       * licensing to implement these interfaces must be obtained separately from Atomikos.
       *
       * Unless required by applicable law or agreed to in writing, software
       * distributed under the License is distributed on an "AS IS" BASIS,
       * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
       */
      
      package com.atomikos.jms.extra;
      
      import java.io.Serializable;
      import java.util.Map;
      
      import javax.jms.Connection;
      import javax.jms.Destination;
      import javax.jms.JMSException;
      import javax.jms.Queue;
      import javax.jms.Session;
      import javax.jms.Topic;
      import javax.transaction.Status;
      import javax.transaction.SystemException;
      
      import com.atomikos.icatch.jta.UserTransactionManager;
      import com.atomikos.icatch.system.Configuration;
      import com.atomikos.jms.AtomikosConnectionFactoryBean;
      import com.atomikos.jms.AtomikosJMSException;
      import com.atomikos.jms.AtomikosTransactionRequiredJMSException;
      
       /**
        * Common functionality for the sender templates.
        *
        */
      
      public abstract class AbstractJmsSenderTemplate 
      {
      
          protected AtomikosConnectionFactoryBean connectionFactoryBean;
          private String user;
          protected String password;
          protected Destination destination;
          private String destinationName;
          private Destination replyToDestination;
          private String replyToDestinationName;
          private int deliveryMode;
          private int priority;
          private long timeToLive;
          protected boolean inited;
      
          protected AbstractJmsSenderTemplate()
          {
              // set default values according to Sun's JMS javadocs
              setTimeToLive ( 0 );
              setDeliveryMode ( javax.jms.DeliveryMode.PERSISTENT );
              setPriority ( 4 );
          }
          
          protected abstract Session getOrRefreshSession ( Connection c ) throws JMSException;
          
          protected abstract Connection getOrReuseConnection() throws JMSException;
          
          protected abstract void afterUseWithoutErrors ( Connection c , Session s ) throws JMSException;
          
          protected void destroy ( Connection c , Session s)
          throws JMSException {
      
              try {
                  if ( s != null ) s.close();
              } catch ( JMSException warn ) {
                  Configuration.logWarning ( this + ": error closing session" , warn);
              }
      
              try {
                  if ( c != null ) c.close();
              } catch ( JMSException warn ) {
                  Configuration.logWarning ( this + ": error closing connection" , warn);
              }
      
          }    
          protected synchronized Connection refreshConnection() throws JMSException {
              Connection connection = null;
              if ( getDestinationName() == null )
                  throw new JMSException ( "Please call setDestination or setDestinationName first!" );
          
              if ( user != null ) {
                  connection = connectionFactoryBean.createConnection (
                          user, password );
          
              } else {
                  connection = connectionFactoryBean.createConnection ();
              }
              connection.start ();
              return connection;
          }
      
      
          /**
           * Initializes the session for sending. 
           * Call this method first.
           */
          
          public void init() throws JMSException
          {
              if ( ! inited ) {
                  if ( connectionFactoryBean == null ) throw new IllegalStateException ( "Property 'atomikosConnectionFactoryBean' must be set first!" );
                  if ( getDestinationName() == null ) {
                      throw new IllegalStateException ( "Property 'destination' or 'destinationName' must be set first!" );
                  }
                  StringBuffer msg = new StringBuffer();
                  msg.append ( this + ":configured with [" );
                  msg.append ( "user=" ).append ( getUser() ).append ( ", " );
                  msg.append ( "password=" ).append ( password ).append ( ", " );
                  msg.append ( "deliveryMode=" ).append ( getDeliveryMode() ).append ( ", " );
                  msg.append ( "timeToLive=" ).append ( getTimeToLive() ).append ( ", " );
                  msg.append ( "priority=" ).append ( getPriority() ).append ( ", " );
                  msg.append ( "destination=" ).append( getDestinationName() ).append ( ", " );
                  msg.append ( "replyToDestination=" ).append ( getReplyToDestinationName() );
                  msg.append ( "]" );
                  if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( msg.toString() );
                  inited = true;        
              }
          }
          
          private void retrieveDestinationIfNecessary() throws JMSException 
          {
              if ( getDestination() == null ) {
                  String dName = getDestinationName();
                  RetrieveDestinationCallback cb = new RetrieveDestinationCallback ( dName );
                  executeCallbackInternal ( cb );
                  setDestination ( cb.getDestination() );
              }
              
          }
          
          private void retrieveReplyToDestinationIfNecessary() throws JMSException 
          {
              if ( getReplyToDestination() == null ) {
                  String dName = getReplyToDestinationName();
                  if ( dName != null ) {
                      RetrieveDestinationCallback cb = new RetrieveDestinationCallback ( dName );
                      executeCallbackInternal ( cb );
                      setReplyToDestination ( cb.getDestination() );
                  }
                  
              }
              
          }
      
          /**
           * Sets the connection factory to use. Required.
           * @param connectionFactory
           */
          public void setAtomikosConnectionFactoryBean(AtomikosConnectionFactoryBean connectionFactory) {
              this.connectionFactoryBean = connectionFactory;
          }
      
          public AtomikosConnectionFactoryBean getAtomikosConnectionFactoryBean() {
              return connectionFactoryBean;
          }
      
          public Destination getDestination() {
              return destination;
          }
          
      
          /**
           * Sets the (provider-specific) destination name in order
           * to lookup the destination (rather than providing one directly).
           * 
           * Required, unless you set the destination directly.
           * 
           * @param destinationName
           */
          
          public void setDestinationName ( String destinationName ) 
          {
              this.destinationName = destinationName;
          }
      
          /**
           * Sets the destination to send to. Required, unless
           * you set the destinationName instead.
           * 
           * @param destination
           */
          public void setDestination(Destination destination) {
              this.destination = destination;
          }
      
          private String getName(Destination d, String destinationName ) {
              String ret = destinationName;
              if ( ret == null ) {
                  if ( d instanceof Queue ) {
                      Queue q = ( Queue ) d;
                      try {
                          ret = q.getQueueName();
                      } catch ( JMSException e ) {
                          if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( this + ": error retrieving queue name" , e );
                      }
                  } else if ( d instanceof Topic ) {
                      Topic t = ( Topic ) d;
                      try {
                          ret = t.getTopicName();
                      } catch ( JMSException e ) {
                          if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( this + ": error retrieving topic name" , e );
                      }
                  }
              }
              return ret;
          }
      
          protected String getDestinationName() {
              return getName ( getDestination() , destinationName );
          }
      
          protected String getReplyToDestinationName() {
              return getName ( getReplyToDestination() , replyToDestinationName );
          }
      
          /**
           * @return The user to connect with, or null if no explicit authentication
           *         is to be used.
           */
          public String getUser() {
              return user;
          }
      
          /**
           * If this session is used for sending request/reply messages, then this
           * property indicates the destination where the replies are to be sent (optional). The
           * session uses this to set the JMSReplyTo header accordingly. This property
           * can be omitted if no reply is needed.
           * 
           * <p>
           * The replyToDestination should be in the same JMS vendor domain as the send
           * queue. To cross domains, configure a bridge for both the request and the
           * reply channels.
           */
          public void setReplyToDestination(Destination destination) 
          {
              this.replyToDestination = destination;
          }
      
          /**
           * Sets the provider-specific replyToDestinationName. Optional.
           * 
           * @param replyToDestinationName
           */
      
          public void setReplyToDestinationName ( String replyToDestinationName ) 
          {
              this.replyToDestinationName = replyToDestinationName;
              
          }    
          
          /**
           * Gets the replyToDestination.
           * 
           * @return
           */
          public Destination getReplyToDestination() {
              return replyToDestination;
          }
      
          /**
           * Set the password for explicit authentication (optional). 
           * This is only required if
           * the user has also been set.
           * 
           * @param password
           *            The password.
           */
          public void setPassword(String password) {
              this.password = password;
          }
      
          /**
           * Set the user to use for explicit authentication (optional). If no explicit
           * authentication is required then this method should not be called.
           * 
           * @param user
           */
          public void setUser(String user) {
              this.user = user;
          }
          
          protected void executeCallbackInternal ( 
                  JmsSenderTemplateCallback callback ) throws JMSException {
              
              init();
              Session  session = null;
              Connection conn = null;
              try {
                  conn = getOrReuseConnection();
                  session = getOrRefreshSession ( conn );    
                  if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( "Calling callback..." );
                  callback.doInJmsSession ( session );
                  if ( Configuration.isDebugLoggingEnabled() ) Configuration.logDebug ( "Callback done!" );
                  afterUseWithoutErrors ( conn , session );
                  
              } catch ( AtomikosTransactionRequiredJMSException notx ) {
                  destroy ( conn , session );
                  String msg = "The JMS session you are using requires a JTA transaction context for the calling thread and none was found." + "
      " +
                  "Please correct your code to do one of the following: " + "
      " +            
                  "1. start a JTA transaction before sending any message, or" + "
      " + 
                  "2. increase the maxPoolSize of the AtomikosConnectionFactoryBean to avoid transaction timeout while waiting for a connection.";
                  Configuration.logWarning ( msg );
                  AtomikosTransactionRequiredJMSException.throwAtomikosTransactionRequiredJMSException ( msg );
          
              } catch ( JMSException e ) {
                  e.printStackTrace();
                  destroy ( conn , session );
                  String msg = this + ": error in sending JMS message";
                  AtomikosJMSException.throwAtomikosJMSException( msg , e );
              }
          }
      
          /**
           * Executes an application-level call-back within the managed session.
           * 
           * @param callback
           * @throws JMSException
           */
          public void executeCallback(JmsSenderTemplateCallback callback) throws JMSException {
              
              init();        
              
      
              retrieveDestinationIfNecessary();
              retrieveReplyToDestinationIfNecessary();
              
              UserTransactionManager tm = new UserTransactionManager ();
              try {
                  if ( tm.getStatus () != Status.STATUS_ACTIVE )
                      throw new JMSException (
                              "This method requires an active transaction!" );
              } catch ( SystemException e ) {
                  Configuration
                          .logWarning ( this +": error in getting transaction status", e );
                  throw new RuntimeException ( e.getMessage () );
              }
              
              executeCallbackInternal ( callback );
              
          }
      
      
      
          /**
           * @return The deliverymode for messages sent in this session.
           */
          public int getDeliveryMode() {
              return deliveryMode;
          }
      
          /**
           * @return The priority for messages sent in this session.
           */
          public int getPriority() {
              return priority;
          }
      
          /**
           * @return The timeToLive for messages sent in this session.
           */
          public long getTimeToLive() {
              return timeToLive;
          }
      
          /**
           * 
           * Set the deliverymode for messages sent in this session (optional). Defaults to
           * persistent.
           * 
           * @param
           */
          public void setDeliveryMode(int i) {
              deliveryMode = i;
          }
      
          /**
           * Set the priority for messages sent in this session (optional). Defaults to 4.
           * 
           * @param
           */
          public void setPriority(int i) {
              priority = i;
          }
      
          /**
           * Set the time to live for messages sent in this session (optional). Defaults to 0.
           * 
           * @param
           */
          public void setTimeToLive(long l) {
              timeToLive = l;
          }
      
          /**
           * Sends a TextMessage.
           * 
           * @param content The text as a string.
           * @throws JMSException 
           */
          public void sendTextMessage(String content) throws JMSException {
              retrieveDestinationIfNecessary();
              retrieveReplyToDestinationIfNecessary();
              SendTextMessageCallback cb = new SendTextMessageCallback ( content , getDestination() , getReplyToDestination() , getDeliveryMode() , getPriority() , getTimeToLive() );
              executeCallback ( cb );
          }
      
          /**
           * Sends a MapMessage.
           * 
           * @param content The Map to get the content from.
           * 
           * @throws JMSException
           */
          public void sendMapMessage(Map content) throws JMSException {
              retrieveDestinationIfNecessary();
              retrieveReplyToDestinationIfNecessary();
              SendMapMessageCallback cb = new SendMapMessageCallback ( content , getDestination() , getReplyToDestination() , getDeliveryMode() , getPriority() , getTimeToLive() );
              executeCallback ( cb );        
          }
      
          /**
           * Sends an ObjectMessage.
           * 
           * @param content The serializable object content.
           * @throws JMSException
           */
          public void sendObjectMessage(Serializable content) throws JMSException {
              retrieveDestinationIfNecessary();
              retrieveReplyToDestinationIfNecessary();
              SendObjectMessageCallback cb = new SendObjectMessageCallback ( content , getDestination() , getReplyToDestination() , getDeliveryMode() , getPriority() , getTimeToLive() );
              executeCallback ( cb );        
          }
      
          /**
           * Sends a ByteMessage.
           * 
           * @param content The content as a byte array.
           * @throws JMSException
           */
          public void sendBytesMessage(byte[] content) throws JMSException {
              retrieveDestinationIfNecessary();
              retrieveReplyToDestinationIfNecessary();
              SendBytesMessageCallback cb = new SendBytesMessageCallback ( content , getDestination() , getReplyToDestination() , getDeliveryMode() , getPriority() , getTimeToLive() );
              executeCallback ( cb );
          }
      
          /**
           * Closes all resources.
           */
          public void close() {
              try {
                  Connection c = getOrReuseConnection();
                  Session s = getOrRefreshSession(c);
                  destroy(c, s);
              } catch (JMSException e) {
                  Configuration.logWarning ( this + ": error closing" , e );
              }
              connectionFactoryBean.close();
          }
      
      }
      复制代码
      复制代码
      /**
       * Copyright (C) 2000-2010 Atomikos <info@atomikos.com>
       *
       * This code ("Atomikos TransactionsEssentials"), by itself,
       * is being distributed under the
       * Apache License, Version 2.0 ("License"), a copy of which may be found at
       * http://www.atomikos.com/licenses/apache-license-2.0.txt .
       * You may not use this file except in compliance with the License.
       *
       * While the License grants certain patent license rights,
       * those patent license rights only extend to the use of
       * Atomikos TransactionsEssentials by itself.
       *
       * This code (Atomikos TransactionsEssentials) contains certain interfaces
       * in package (namespace) com.atomikos.icatch
       * (including com.atomikos.icatch.Participant) which, if implemented, may
       * infringe one or more patents held by Atomikos.
       * It should be appreciated that you may NOT implement such interfaces;
       * licensing to implement these interfaces must be obtained separately from Atomikos.
       *
       * Unless required by applicable law or agreed to in writing, software
       * distributed under the License is distributed on an "AS IS" BASIS,
       * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
       */
      
      package com.atomikos.jms.extra;
      
      import java.util.ArrayList;
      import java.util.Iterator;
      import java.util.List;
      
      import javax.jms.Destination;
      import javax.jms.ExceptionListener;
      import javax.jms.JMSException;
      import javax.jms.MessageListener;
      
      import com.atomikos.icatch.system.Configuration;
      import com.atomikos.jms.AtomikosConnectionFactoryBean;
      
       /**
        * 
        * A message-driven container for asynchronously receiving JMS messages
        * from a topic or queue, within a managed JTA transaction context.
        * 
        * Upon start, an instance of this class will create a number of
        * concurrent sessions that listen for incoming messages on the same destination.
        * MessageListener instances should be thread-safe if the pool size is larger
        * than one. Note: in general, after start() any changed properties are only
        * effective on the next start() event.
        * 
        * <p>
        * <b>IMPORTANT:</b> the transactional behaviour guarantees redelivery after failures.
        * As a side-effect, this can lead to so-called <em>poison messages</em>: messages
        * whose processing repeatedly fails due to some recurring error (for instance, a primary
        * key violation in the database, a NullPointerException, ...). Poison messages are problematic
        * because they can prevent other messages from being processed, and block the system.
        *  
        * To avoid poison messages, make sure that your MessageListener implementation 
        * only throws a <b>RuntimeException</b> when the problem is <em>transient</em>. In that
        * case, the system will perform rollback and the message will be redelivered
        * facing a clean system state. All non-transient errors (i.e., those that happen
        * each time a message is delivered) indicate problems at the application level
        * and should be dealt with by writing better application code.
        */
      
      public class MessageDrivenContainer 
      implements MessageConsumerSessionProperties
      {
          private static final int DEFAULT_TIMEOUT = 30;
      
          
          private AtomikosConnectionFactoryBean connectionFactoryBean;
          private MessageListener messageListener;
          private String user;
          private String password;
          private Destination destination;
          private String destinationName;
          private int transactionTimeout;
          private int poolSize;
          private List sessions;
          private boolean daemonThreads;
          private boolean notifyListenerOnClose;
          private String messageSelector;
          private ExceptionListener exceptionListener;
          private String subscriberName;
          private boolean noLocal;
          private boolean unsubscribeOnClose;
          private String clientID;
          
          private int receiveTimeout;
          
          public MessageDrivenContainer()
          {
              sessions = new ArrayList ();
              notifyListenerOnClose = false;
              setPoolSize ( 1 );
              setTransactionTimeout ( DEFAULT_TIMEOUT );
          }
          
          private MessageConsumerSession createSession() 
          {
              return new MessageConsumerSession ( this );
          }
          
          /**
           * Sets the clientID for durable subscriptions. Optional.
           * 
           * @param clientID
           */
          public void setClientID ( String clientID ) {
              this.clientID = clientID;
          }
          
          
          
          /**
           * Sets the connection factory to use. Required.
           * @param bean
           */
          public void setAtomikosConnectionFactoryBean ( AtomikosConnectionFactoryBean bean )
          {
              this.connectionFactoryBean = bean;
          }
          
          public AtomikosConnectionFactoryBean getAtomikosConnectionFactoryBean()
          {
              return connectionFactoryBean;
          }
          
          /**
           * Gets the destination. 
           * 
           * @return The destination, or null if not set.
           */
          public Destination getDestination()
          {
              return destination;
          }
          
          /**
           * Sets the JMS destination to listen on (required unless the destinationName is set instead).
           * 
           * @param dest
           */
          public void setDestination ( Destination dest )
          {
              this.destination = dest;
          }
          
          /**
           * Gets the destination name.
           * 
           * @return The name, or null if not set.
           */
          public String getDestinationName()
          {
              return destinationName;
          }
          
          /**
           * Sets the JMS provider-specific destination name
           * (required unless the destination is set directly). 
           * 
           * @param destinationName
           */
          public void setDestinationName ( String destinationName )
          {
              this.destinationName = destinationName;
          }
          
          
          /**
           * Sets whether threads should be daemon threads or not (optional).
           * Default is false.
           * @param value If true then threads will be daemon threads.
           */
          public void setDaemonThreads ( boolean value ) 
          {
                  this.daemonThreads = value;
          }
      
          /**
           * Tests whether threads are daemon threads.
           * @return True if threads are deamons.
           */
          public boolean getDaemonThreads() 
          {
                  return daemonThreads;
          }
      
          /**
           * 
           * Get the message listener if any.
           * 
           * @return
           */
          public MessageListener getMessageListener() 
          {
              return messageListener;
          }
      
          /**
           * Get the transaction timeout.
           * 
           * @return
           */
          public int getTransactionTimeout() 
          {
              return transactionTimeout;
          }
      
          /**
           * Get the user for connecting, or null if the default user should be used.
           * 
           * @return
           */
          public String getUser() 
          {
              return user;
          }
      
          /**
           * Set the message listener to use (required). 
           * The same instance will be used for each
           * session in the pool, meaning that instances need to be thread-safe. Only
           * one listener is allowed at a time. Call this method with a null argument
           * to unset the listener.
           * 
           * @param listener
           */
          public void setMessageListener ( MessageListener listener ) 
          {
          
              messageListener = listener;
              Iterator it = sessions.iterator ();
              while ( it.hasNext () ) {
                  MessageConsumerSession s = (MessageConsumerSession) it.next ();
                  s.setMessageListener ( listener );
              }
          }
      
          /**
           * Set the password if explicit authentication is needed (optional). 
           * You need to set this if the user is also set.
           * 
           * @param string
           */
          public void setPassword ( String string ) 
          {
              password = string;
          }
      
          /**
           * Set the transaction timeout in seconds (optional).
           * 
           * @param seconds
           */
          public void setTransactionTimeout ( int seconds ) 
          {
              transactionTimeout = seconds;
          }
      
          /**
           * Set the user to use for explicit authentication (optional). 
           * Don't set this property
           * if you want to use the default authentication.
           * 
           * @param string
           */
          public void setUser ( String string ) 
          {
              user = string;
          }
      
          /**
           * Get the message selector (if any)
           * 
           * @return The selector, or null if none.
           */
          public String getMessageSelector() 
          {
              return this.messageSelector;
          }
      
          /**
           * Set the message selector to use (optional).
           * 
           * @param selector
           */
          public void setMessageSelector ( String selector ) 
          {
              this.messageSelector = selector;
          }
      
          /**
           * Get the size of the pool.
           * 
           * @return
           */
          public int getPoolSize() 
          {
              return poolSize;
          }
      
          /**
           * Sets the size of the session pool (optional).
           * Default is 1.
           * 
           * @param size 
           */
          public void setPoolSize ( int size ) 
          {
              poolSize = size;
          }
      
          /**
           * Gets the exception listener (if any). 
           * @return Null if no ExceptionListener was set.
           */
          public ExceptionListener getExceptionListener() 
          {
              return exceptionListener;
          }
      
          /**
           * Sets the exception listener (optional). The listener will be
           * notified of connection-level JMS errors.
           * 
           * @param exceptionListener
           */
          public void setExceptionListener ( ExceptionListener exceptionListener ) 
          {
              this.exceptionListener = exceptionListener;
          }
          
          /**
           * Test if this instance will receive sends from the same connection.
           * 
           * @return
           */
          public boolean isNoLocal() {
              return noLocal;
          }
          
          /**
           * Sets whether or not this topic should receive sends from the 
           * same connection (optional). 
           * 
           * @param noLocal 
           */
      
          public void setNoLocal(boolean noLocal) {
              this.noLocal = noLocal;
          }
      
          /**
           * Gets the subscriber name (for durable subscribers).
           * @return The name, or null if not set (no durable subscriber).
           */
          
          public String getSubscriberName() {
              return subscriberName;
          }
      
          /**
           * Sets the name to use for durable subscriptions (optional).
           * <br>
           * <b>Note: this name will be appended with a suffix to ensure uniqueness
           * among instances in the pool. Otherwise, the JMS back-end would see 
           * multiple instances subscribing with the same name - an error.</b>
           * 
           * @param subscriberName
           */
          
          public void setSubscriberName(String subscriberName) {
              this.subscriberName = subscriberName;
          }
      
          protected boolean getNoLocal() {
              
              return isNoLocal();
          }
          
          /**
           * Start listening for messages.
           * 
           * @throws JMSException
           */
          public void start() throws JMSException 
          {
              if ( destination == null && destinationName == null )
                  throw new JMSException (
                          "MessageDrivenContainer: destination not specified" );
              if ( connectionFactoryBean == null )
                  throw new JMSException (
                          "MessageDrivenContainer: factory not set" );
              if ( messageListener == null )
                  throw new JMSException (
                          "MessageDrivenContainer: messageListener not set" );
              for ( int i = 0; i < poolSize; i++ ) {
                  MessageConsumerSession s = createSession();
                  s.setMessageListener ( messageListener );
                  s.setPassword ( password );
                  s.setUser ( user );
                  s.setDestination ( destination );
                  s.setDestinationName ( destinationName );
                  s.setAtomikosConnectionFactoryBean ( connectionFactoryBean );
                  s.setDaemonThreads ( daemonThreads );
                  s.setNotifyListenerOnClose ( notifyListenerOnClose );
                  s.setMessageSelector ( getMessageSelector () );
                  s.setExceptionListener ( exceptionListener );
                  s.setNoLocal( noLocal );
                  s.setSubscriberName( subscriberName );
                  //set subscriber name with suffix to ensure unique names
                  if ( getSubscriberName() != null ) s.setSubscriberName ( getSubscriberName() + "-" + i );
                  s.setNoLocal ( getNoLocal() );
                  s.setClientID(clientID);
                  try {
                      s.startListening ();
                      // System.out.println ( "MessageDrivenContainer: started
                      // session");
                  } catch ( Exception e ) {
                      Configuration.logWarning ( "Error starting pool", e );
                  }
                  sessions.add ( s );
              }
          
              // set listener again to trigger listening
              setMessageListener ( messageListener );
          }
      
          /**
           * Stop listening for messages. If <b>notifyListenerOnClose</b> is set then
           * calling this method will notify the listener by calling its onMessage
           * method with a null argument (and also without transaction context).
           * 
           * This method will wait for all active receive operations to unblock, which may take
           * up to <b>receiveTimeout</b> seconds per active thread.
           */
          public void stop() 
          {
              Iterator it = sessions.iterator ();
              while ( it.hasNext () ) {
                  MessageConsumerSession s = (MessageConsumerSession) it.next ();
                  s.stopListening ();
              }
          }
      
          /**
           * Getter to check whether the listener is notified on close.
           * 
           * @return
           */
          public boolean getNotifyListenerOnClose() 
          {
              return notifyListenerOnClose;
          }
      
          /**
           * Set whether the listener should be notified of close events on the pool
           * (optional). Default is false.
           * 
           * @param b
           *            If true, then the listener will receive a null message if the
           *            pool is closed.
           */
          public void setNotifyListenerOnClose ( boolean b ) 
          {
              notifyListenerOnClose = b;
              Iterator it = sessions.iterator ();
              while ( it.hasNext () ) {
                    MessageConsumerSession s = (MessageConsumerSession) it.next ();
                  s.setNotifyListenerOnClose ( b );
              }
          }
          
          /**
           * Sets whether unsubscribe should be done at closing time (optional). Default is false.
           * 
           * @param b If true, then unsubscribe will be done at closing time. This only applies to 
           * durable subscribers (i.e., cases where subscriberName is set).
           */
          public void setUnsubscribeOnClose ( boolean b ) 
          {
              this.unsubscribeOnClose = b;
          }
          
          /**
           * Getter to test if unsubscribe should be called on close.
           */
          
          public boolean getUnsubscribeOnClose()
          {
              return unsubscribeOnClose;
          }
          
      
          /**
           * Gets the receive timeout in seconds.
           * 
           * @return
           */
          public int getReceiveTimeout() {
              int ret = receiveTimeout;
              if ( ret <=0 ) ret = getTransactionTimeout()/2;
              return ret;
          }
      
          
          /**
           * Sets the receive timeout in seconds, 
           * i.e. the number of seconds to wait for incoming messages in the message listener thread's event loop.
           *  
           * This property is optional and defaults to half the transactionTimeout, but typically this should be lower 
           * because the time required to shutdown (stop) this container will be bound by this value multiplied by 
           * the number of threads (as indicated by <b>poolSize</b>).
           * 
           * @param seconds
           */
          public void setReceiveTimeout(int seconds) {
              this.receiveTimeout = seconds;    
          }
      
      }
      复制代码
    6. 请检查我们完整的用户指南

    注意:虽然这个版本包含了特定的第三方产品,如RDBMS软件和JMS代理软件,请注意,AtomikosTransactionsEssentials绝不是仅限制在这些特定的产品,除了RDBMS软件和JMS代理软件大多数的其他软件依然兼容。

    J2SE实例:examples/jse这个文件夹包含带有源代码的各种各样的例子。可以执行这个脚本来运行这个程序(示例程序只有文本输出,而没有图形化界面输出)

             运行示例前最好安装Ant,然后打开命令窗口在 examples/jse 文件夹并且输入“ant”。

    Linux/Unix/Mac OSX Note在这些系统,你必须在终端输入'chmod u+x *.sh'命令,否则,示例将无法运行。

    解决问题:偶尔,一些示例可能不能再继续启动使用(尽管在开始阶段可以正常运行),这个通常是由于进程原因引起的(比如JMS代理后台环境),关闭这期间的所有进程便会解决这个问题,或者,重新启动计算机,也可以解决这个问题。

    在Spring中使用:Atomikos TransactionEssentials可以很方便的于Spring相结合,运行您程序使用企业级的J2EE应用程序,而不需要EJB甚至EJB容器。另外,我们额外提供了一个强大的功能,JMS消息驱动添加到Spring的内置特性。

             查看examples/spring文件夹,将展示Atomikos TransactionEssentials如何配置基于Spring的应用程序。

    在J2EE的web容器中使用:在您的J2EE应用程序环境中使用Atomikos TransactionsEssentials最简单的方法是通过Spring作为pico-container(组件容器),如果想获取更多信息,请到Spring处查看。

    Javadoc

    http://www.atomikos.com/downloads/transactions-essentials/com/atomikos/AtomikosTransactionsEssentials/javadoc/3.7/index.html

  • 相关阅读:
    zabbix(x)
    kvm 学习(三)存储池
    kvm 学习(二)镜像
    hadoop3.1.1:找不到或无法加载主类 org.apache.hadoop.mapreduce.v2.app.MRAppMaster
    (转)mysql更改数据目录
    (转)SLOW READPROCESSOR;ERROR SLOW BLOCKRECEIVER错误日志分析
    Linux下 为什么有时候使用sudo也提示没有权限
    (转)hadoop 常规错误问题(一)
    (转)SmartPing:一个服务器Ping值监测工具
    (转)hadoop 配置文件解释
  • 原文地址:https://www.cnblogs.com/duanxz/p/4672551.html
Copyright © 2011-2022 走看看