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,只需要按以下步驟
- 將idst和lib中的jar包全部放入的項目中
- 創建或者自定義你應用的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 - 在你的應用程序中，創建一個實例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; } }
- 對於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; } }
- 對於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; } }
- 請檢查我們完整的用戶指南
注意：雖然這個版本包含了特定的第三方產品，如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