Atomikos 中文说明文档(周枫翻译)

时间:2022-08-25 05:42:41

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." + "\n" +
    "Please correct your code to do one of the following: " + "\n" +
    "1. start a JTA transaction before sending any message, or" + "\n" +
    "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