ByteTCC Transaction Manager旨在提供一个兼容JTA的基于TCC机制的分布式事务管理器

/** * Copyright 2014-2016 yangming.liu<bytefox@126.com>. * * This copyrighted material is made available to anyone wishing to use, modify, * copy, or redistribute it subject to the terms and conditions of the GNU * Lesser General Public License, as published by the Free Software Foundation. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License * for more details. * * You should have received a copy of the GNU Lesser General Public License * along with this distribution; if not, see <http://www.gnu.org/licenses/>. */ package org.bytesoft.bytetcc; import java.io.Serializable; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import javax.transaction.HeuristicMixedException; import javax.transaction.HeuristicRollbackException; import javax.transaction.RollbackException; import javax.transaction.Status; import javax.transaction.Synchronization; import javax.transaction.SystemException; import javax.transaction.xa.XAException; import javax.transaction.xa.XAResource; import javax.transaction.xa.Xid; import org.apache.commons.lang3.StringUtils; import org.bytesoft.bytejta.supports.jdbc.RecoveredResource; import org.bytesoft.bytejta.supports.resource.RemoteResourceDescriptor; import org.bytesoft.bytetcc.supports.CompensableRolledbackMarker; import org.bytesoft.bytetcc.supports.resource.LocalResourceCleaner; import org.bytesoft.common.utils.ByteUtils; import org.bytesoft.common.utils.CommonUtils; import org.bytesoft.compensable.CompensableBeanFactory; import org.bytesoft.compensable.CompensableInvocation; import org.bytesoft.compensable.CompensableTransaction; import org.bytesoft.compensable.ContainerContext; import org.bytesoft.compensable.TransactionContext; import org.bytesoft.compensable.archive.CompensableArchive; import org.bytesoft.compensable.archive.TransactionArchive; import org.bytesoft.compensable.logging.CompensableLogger; import org.bytesoft.transaction.CommitRequiredException; import org.bytesoft.transaction.RollbackRequiredException; import org.bytesoft.transaction.Transaction; import org.bytesoft.transaction.TransactionRepository; import org.bytesoft.transaction.archive.XAResourceArchive; import org.bytesoft.transaction.remote.RemoteCoordinator; import org.bytesoft.transaction.remote.RemoteNode; import org.bytesoft.transaction.remote.RemoteSvc; import org.bytesoft.transaction.supports.TransactionExtra; import org.bytesoft.transaction.supports.TransactionListener; import org.bytesoft.transaction.supports.TransactionListenerAdapter; import org.bytesoft.transaction.supports.TransactionResourceListener; import org.bytesoft.transaction.supports.resource.XAResourceDescriptor; import org.bytesoft.transaction.supports.serialize.XAResourceDeserializer; import org.bytesoft.transaction.xa.TransactionXid; import org.bytesoft.transaction.xa.XidFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class CompensableTransactionImpl extends TransactionListenerAdapter implements CompensableTransaction, CompensableRolledbackMarker { static final Logger logger = LoggerFactory.getLogger(CompensableTransactionImpl.class); private final TransactionContext transactionContext; private final List<CompensableArchive> archiveList = new ArrayList<CompensableArchive>(); private final Map<RemoteSvc, XAResourceArchive> resourceMap = new HashMap<RemoteSvc, XAResourceArchive>(); private final List<XAResourceArchive> resourceList = new ArrayList<XAResourceArchive>(); private final Map<Thread, Transaction> transactionMap = new ConcurrentHashMap<Thread, Transaction>(); private CompensableBeanFactory beanFactory; private int transactionVote; private int transactionStatus = Status.STATUS_ACTIVE; /* current compensable-decision in confirm/cancel phase. */ private transient Boolean positive; /* current compensable-archive in confirm/cancel phase. */ private transient CompensableArchive archive; private transient final Map<Xid, List<CompensableArchive>> xidToArchivesMap = new HashMap<Xid, List<CompensableArchive>>(); private transient final Map<Xid, TransactionBranch> xidToBranchMap = new HashMap<Xid, TransactionBranch>(); private Map<String, Serializable> variables = new HashMap<String, Serializable>(); private Thread currentThread; private final Lock lock = new ReentrantLock(); private transient Exception createdAt; public CompensableTransactionImpl(TransactionContext txContext) { this.transactionContext = txContext; } public TransactionArchive getTransactionArchive() { TransactionArchive transactionArchive = new TransactionArchive(); transactionArchive.setVariables(this.variables); transactionArchive.setCoordinator(this.transactionContext.isCoordinator()); transactionArchive.setPropagated(this.transactionContext.isPropagated()); transactionArchive.setCompensable(this.transactionContext.isCompensable()); transactionArchive.setCompensableStatus(this.transactionStatus); transactionArchive.setVote(this.transactionVote); transactionArchive.setXid(this.transactionContext.getXid()); transactionArchive.getRemoteResources().addAll(this.resourceList); transactionArchive.getCompensableResourceList().addAll(this.archiveList); transactionArchive.setPropagatedBy(this.transactionContext.getPropagatedBy()); transactionArchive.setRecoveredAt(this.transactionContext.getCreatedTime()); transactionArchive.setRecoveredTimes(this.transactionContext.getRecoveredTimes()); return transactionArchive; } public synchronized void participantCommit(boolean opc) throws RollbackException, HeuristicMixedException, HeuristicRollbackException, SecurityException, IllegalStateException, CommitRequiredException, SystemException { // Recover if transaction is recovered from tx-log. this.recoverIfNecessary(); if (this.transactionStatus != Status.STATUS_COMMITTED) { this.fireCommit(); // TODO } } public synchronized void commit() throws RollbackException, HeuristicMixedException, HeuristicRollbackException, SecurityException, IllegalStateException, SystemException { if (this.transactionStatus == Status.STATUS_ACTIVE) { this.fireCommit(); } else if (this.transactionStatus == Status.STATUS_MARKED_ROLLBACK) { this.fireRollback(); throw new HeuristicRollbackException(); } else if (this.transactionStatus == Status.STATUS_ROLLEDBACK) /* should never happen */ { throw new RollbackException(); } else if (this.transactionStatus == Status.STATUS_COMMITTED) /* should never happen */ { logger.debug("Current transaction has already been committed."); } else { throw new IllegalStateException(); } } private void fireCommit() throws RollbackException, HeuristicMixedException, HeuristicRollbackException, SecurityException, IllegalStateException, SystemException { CompensableLogger compensableLogger = this.beanFactory.getCompensableLogger(); this.transactionContext.setCompensating(true); this.transactionStatus = Status.STATUS_COMMITTING; compensableLogger.updateTransactionStatus(this.getTransactionArchive()); // compensableLogger.updateTransaction(this.getTransactionArchive()); SystemException systemEx = null; try { this.fireNativeParticipantConfirm(); } catch (SystemException ex) { systemEx = ex; logger.info("{}| confirm native branchs failed!", ByteUtils.byteArrayToString(this.transactionContext.getXid().getGlobalTransactionId()), ex); } catch (RuntimeException ex) { systemEx = new SystemException(XAException.XAER_RMERR); systemEx.initCause(ex); logger.info("{}| confirm native branchs failed!", ByteUtils.byteArrayToString(