忧郁者 发表于 2017-2-19 07:57:01

Ehcache 事务管理源码探析

  可能与大家关注点有不同,有考虑不周处,请大家指出...
  Ehcache获取分布式事务支持可从net.sf.ehcache.transaction.manager.DefaultTransactionManagerLookup类中知晓:



private final JndiSelector defaultJndiSelector = new JndiSelector("genericJNDI", "java:/TransactionManager");
    private final Selector[] transactionManagerSelectors = new Selector[] {defaultJndiSelector,
            new JndiSelector("Weblogic", "javax.transaction.TransactionManager"),
            new FactorySelector("Bitronix", "bitronix.tm.TransactionManagerServices"),
            new ClassSelector("Atomikos", "com.atomikos.icatch.jta.UserTransactionManager")};
  默认获取JNDI名“java:/TransactionManager”。JBoss JTA事务。如果需要其他类型的事务管理获取方式,
  通过配置TransactionManagerLookup来替代默认的。
  配置如下:


  <transactionManagerLookup
class= "com.mycompany.transaction.manager.MyTransactionManagerLookupClass"
properties="" propertySeparator=":"/>
  另外当使用默认的事务管理查找类,但是不同的JNDI查找名称,可以配置:


  <transactionManagerLookup
class="net.sf.ehcache.transaction.manager.DefaultTransactionManagerLookup"
properties="jndiName=java:appserver/TransactionManager" propertySeparator=";"/>
  现跟踪Ehcache PUT操作时是如何加入事务的。
  当Ehcache配置成xa或者xa-strict时,内部使用net.sf.ehcache.transaction.xa.XATransactionStore存储逻辑,put操作如下:



public boolean put(Element element) throws CacheException {
      LOG.debug("cache {} put {}", cache.getName(), element);
      // this forces enlistment so the XA transaction timeout can be propagated to the XA resource
      getOrCreateTransactionContext();
      Element oldElement = getQuietFromUnderlyingStore(element.getObjectKey());
      return internalPut(new StorePutCommand(oldElement, copyElementForWrite(element)));
    }
  红色部分是事务关键操作:初始化事务上下文。事务上下文维护着该事务的所有cache操作、remove操作的keys、put操作的keys和underlyingStore。



    private XATransactionContext getOrCreateTransactionContext() {
      try {
            EhcacheXAResourceImpl xaResource = getOrCreateXAResource();
            XATransactionContext transactionContext = xaResource.getCurrentTransactionContext();
            if (transactionContext == null) {
                transactionManagerLookup.register(xaResource);
                LOG.debug("creating new XA context");
                transactionContext = xaResource.createTransactionContext();
                xaResource.addTwoPcExecutionListener(new UnregisterXAResource());
            } else {
                transactionContext = xaResource.getCurrentTransactionContext();
            }
            LOG.debug("using XA context {}", transactionContext);
            return transactionContext;
      } catch (SystemException e) {
            throw new TransactionException("cannot get the current transaction", e);
      } catch (RollbackException e) {
            throw new TransactionException("transaction rolled back", e);
      }
    }

构建事务上下文的操作:




public XATransactionContext createTransactionContext() throws SystemException, RollbackException {
      XATransactionContext ctx = getCurrentTransactionContext();
      if (ctx != null) {
            return ctx;
      }
      Transaction transaction = txnManager.getTransaction();
      LOG.debug("enlisting {} in {}", this, transaction);
      transaction.enlistResource(this);//关键操作:注册当前XAResource到事务中
      // currentXid is set by a call to start() which itself is called by transaction.enlistResource(this)
      if (currentXid == null) {
            throw new CacheException("enlistment of XAResource of cache named '" + getCacheName() +
                  "' did not end up calling XAResource.start()");
      }
      ctx = xidToContextMap.get(currentXid);
      if (ctx == null) {
            LOG.debug("creating new context for XID [{}]", currentXid);
            ctx = new XATransactionContext(underlyingStore);
            xidToContextMap.put(currentXid, ctx);
      }
      return ctx;
    }

net.sf.ehcache.transaction.xa.EhcacheXAResourceImpl为Ehcache XAResource的实现。

研究一下XAResource都干了些啥。无非是该事务源相关属性及提交、回滚等操作吧。具体看一下实现代码:

关键属性:




    private final Ehcache cache;
    private final Store underlyingStore;
    private final TransactionIDFactory transactionIDFactory;
    private final TransactionManager txnManager;
    private final SoftLockFactory softLockFactory;
    private final ConcurrentMap<Xid, XATransactionContext> xidToContextMap = new ConcurrentHashMap<Xid, XATransactionContext>();
    private final XARequestProcessor processor;
    private volatile Xid currentXid;
    private volatile int transactionTimeout;
    private final List<XAExecutionListener> listeners = new ArrayList<XAExecutionListener>();
    private final ElementValueComparator comparator;
......

关键方法:




public void commit(Xid xid, boolean onePhase)
public void rollback(Xid xid) throws XAException
/**
   * Add a listener which will be called back according to the 2PC lifecycle
   * @param listener the XAExecutionListener
   */
void addTwoPcExecutionListener(XAExecutionListener listener);
/**
   * Obtain the already associated {@link XATransactionContext} with the current Transaction,
   * or create a new one should none be there yet.
   * @return The associated Transaction associated {@link XATransactionContext}
   */
    XATransactionContext createTransactionContext() throws SystemException, RollbackException;

前两个为XAResource接口抽象方法,完成事务的基本操作,在JTA事务提交或是回滚时会被调用;后面是Ehcache抽象出来的接口方法。

接着关心这里的rollback()方法都做了啥。关键代码如下:




int rc = prepareInternal(xid);
  if (rc == XA_RDONLY) {
   return;
   }





    public int prepareInternal(Xid xid) throws XAException {
      fireBeforePrepare();
      XATransactionContext twopcTransactionContext = xidToContextMap.get(xid);
      if (twopcTransactionContext == null) {
            throw new EhcacheXAException("transaction never started: " + xid, XAException.XAER_NOTA);
      }

      XidTransactionID xidTransactionID = transactionIDFactory.createXidTransactionID(xid);
      List<Command> commands = twopcTransactionContext.getCommands();
      List<Command> preparedCommands = new LinkedList<Command>();
      boolean prepareUpdated = false;
      LOG.debug("preparing {} command(s) for [{}]", commands.size(), xid);
      for (Command command : commands) {
            try {
                prepareUpdated |= command.prepare(underlyingStore, softLockFactory, xidTransactionID, comparator);
                preparedCommands.add(0, command);
            } catch (OptimisticLockFailureException ie) {
                for (Command preparedCommand : preparedCommands) {
                  preparedCommand.rollback(underlyingStore);
                }
                preparedCommands.clear();
                throw new EhcacheXAException(command + " failed because value changed between execution and 2PC",
                        XAException.XA_RBINTEGRITY, ie);
            }
      }
      xidToContextMap.remove(xid);
      if (!prepareUpdated) {
            rollbackInternal(xid);
      }
      LOG.debug("prepared xid [{}] read only? {}", xid, !prepareUpdated);
      return prepareUpdated ? XA_OK : XA_RDONLY;
    }

可以看到进行了底部存储逻辑的回滚处理。

实例化好XAResource后回到初始化事务上下文方法getOrCreateTransactionContext中,可以知道通过该XAResource直接取得XATransactionContext。并且给XAResource注册了两个监听器UnregisterXAResource和CleanupXAResource(在事务提交或是混回滚时启动)。

至此事务的初始化工作完成。



理一下put过程。外部调用XATransactionStore的put方法,向当前XID的XATransactionContext中添加addCommand(封装了针对指定cache具体的提交与回滚操作),接着控制权到事务管理中,事务管理器管理EhcacheXAResourceImpl对象,提交与回滚操作通过取得XATransactionContext中的Commands对象并执行来完成。  欢迎大家批评指正!
页: [1]
查看完整版本: Ehcache 事务管理源码探析