Java 类org.springframework.transaction.support.TransactionSynchronizationManager 实例源码

private void handleComposeMessageCreateCommand(final Object obj) throws IOException {
  final ComposeMessageCreateCommand cmd = (ComposeMessageCreateCommand) obj;
  final Message message = new MessageBuilder().transferId(cmd.getTransferId())
      .messageState(MessageState.TO_COMPOSE).build();


  message.setInternalData(InternalDataUtils.convertInternalDataToJson(cmd.getData()));

  final BasicOutboundConfiguration configuration =
      basicConfigurationRepository.findOne(cmd.getConfigId());
  message.setOutboundConfiguration(configuration);

  messageRepository.save((MessageImpl) message);
  final String actorId = getContext().parent().path().name();
  TransactionSynchronizationManager
      .registerSynchronization(new TransactionSynchronizationAdapter() {
        @Override
        public void afterCommit() {
          LOG.info("Saved new [{}]", message);
          getSender().tell(new ComposeMessageCreatedEvent(actorId, message.getId()), getSelf());
          stop();
        }
      });
}

@Override
public void afterCompletion(WebRequest request, Exception ex) throws DataAccessException {
    String participateAttributeName = getParticipateAttributeName();
    Integer count = (Integer) request.getAttribute(participateAttributeName, WebRequest.SCOPE_REQUEST);
    if (count != null) {
        // Do not modify the PersistenceManager: just clear the marker.
        if (count > 1) {
            request.setAttribute(participateAttributeName, count - 1, WebRequest.SCOPE_REQUEST);
        }
        else {
            request.removeAttribute(participateAttributeName, WebRequest.SCOPE_REQUEST);
        }
    }
    else {
        PersistenceManagerHolder pmHolder = (PersistenceManagerHolder)
                TransactionSynchronizationManager.unbindResource(getPersistenceManagerFactory());
        logger.debug("Closing JDO PersistenceManager in OpenPersistenceManagerInViewInterceptor");
        PersistenceManagerFactoryUtils.releasePersistenceManager(
                pmHolder.getPersistenceManager(), getPersistenceManagerFactory());
    }
}

/**
 * <strong>此方法仅限与bdf2-core模块配合使用,且只能用于不加事务的方法当中，如果当前方法或当前方便的调用方法需要用到事务，
 * 一定不能使用该方法，否则事务将不起作用</strong><br>
 * 此方法返回当前线程当中指定的数据源对应的Session,如线程中未指定，则返回默认的Session,<br>
 * 注意此方法返回的Session对象我们在使用完毕之后，无需将其close，<br>
 * 在线程执行完毕时，BDF会自动将此方法产生的Session对象进行close处理；同时，此方法仅限在标准的WEB请求调用时使用，<br>
 * 如果在后台的JOB中调用此方法可能会有异常抛出
 * 
 * @return 返回一个标准的Hibernate Session对象
 */
public Session getSession() {
    String dsName = sessionFactoryRepository.getDefaultSessionFactoryName();
    dsName = this.getDataSourceName(dsName);
    if (StringUtils.isEmpty(TransactionSynchronizationManager.getCurrentTransactionName())) {
        Map<String, Session> map = ContextHolder.getHibernateSessionMap();
        if (map == null) {
            throw new RuntimeException("This method can only used in bdf2-core module");
        }
        if (map.containsKey(dsName) && map.get(dsName).isOpen()) {
            return map.get(dsName);
        } else {
            if (map.containsKey(dsName)) {
                map.remove(dsName);
            }
            Session session = getSessionFactory().openSession();
            map.put(dsName, session);
            return session;
        }
    } else {
        return this.getSessionFactory(dsName).getCurrentSession();
    }
}

public boolean startNewThread(ActivityRunner runner) {
    boolean wasStarted = false;
    if (isStarted()) {
        synchronized (runningActivities) {
            Assert.isFalse(runningActivities.contains(runner.getActivityId()), "Error");
            runningActivities.add(runner.getActivityId());
            connectionCapManager.add(runner.getActivityName(), runner.getActivityId());

            // TX Synchro
            try {
                final ThreadStarterSynchronization synchro = new ThreadStarterSynchronization(runner);
                TransactionSynchronizationManager.registerSynchronization(synchro);
                wasStarted = true;
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        if (runner.isTopActivity()) {
            synchronized (runningTopActivities) {
                runningTopActivities.add(runner.getActivityId());
            }
        }
    }
    return wasStarted;
}

@Override
public Object intercept(Invocation invocation) throws Throwable {

    boolean synchronizationActive = TransactionSynchronizationManager.isSynchronizationActive();
    //如果没有事务
    if (!synchronizationActive) {
        Object[] objects = invocation.getArgs();
        MappedStatement ms = (MappedStatement) objects[0];

        DynamicDataSourceGlobal dynamicDataSourceGlobal;

        if ((dynamicDataSourceGlobal = CACHE_MAP.get(ms.getId())) == null) {
            dynamicDataSourceGlobal = getDynamicDataSource(ms, objects[1]);
            LOGGER.warn("设置方法[{}] use [{}] Strategy, SqlCommandType [{}]..", ms.getId(), dynamicDataSourceGlobal.name(), ms.getSqlCommandType().name());
            CACHE_MAP.put(ms.getId(), dynamicDataSourceGlobal);
        }
        DynamicDataSourceHolder.putDataSource(dynamicDataSourceGlobal);
    }

    return invocation.proceed();
}

/**
 * Get the transaction identifier used to store it in the transaction map.
 * 
 * @param tx Transaction
 * @param storeRef StoreRef
 * @return - the transaction id
 */
@SuppressWarnings("unchecked")
private String getTransactionId(Transaction tx, StoreRef storeRef)
{
    if (tx instanceof SimpleTransaction)
    {
        SimpleTransaction simpleTx = (SimpleTransaction) tx;
        return simpleTx.getGUID();
    }
    else if (TransactionSynchronizationManager.isSynchronizationActive())
    {
        Map<StoreRef, LuceneIndexer> indexers = (Map<StoreRef, LuceneIndexer>) AlfrescoTransactionSupport.getResource(indexersKey);
        if (indexers != null)
        {
            LuceneIndexer indexer = indexers.get(storeRef);
            if (indexer != null)
            {
                return indexer.getDeltaId();
            }
        }
    }
    return null;
}

@PostUpdate
@Async
public void postUpdate(Object object) {
    LOG.info("Listening to post update for object:" + object);
    // Entitys have to be annotated with @EventListeners and reference this class in that annotation, because of this
    // the usages of this class are not executed withing the handle of the Spring context. So now we have to use this funky
    // ass way of wiring in fields AS this method is being called. #sadface
    AutowireHelper.autowire(this);
    // Trying to just add @Transactional(Transactional.TxType.REQUIRES_NEW) to this method didn't work at all, it was just being ignored.
    // This wrapper is what ended up working.
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
        @Override
        public void afterCompletion(int status) {
            super.afterCompletion(status);
            List<Webhook> hooks = webhookManager.retrieveWebhooksByEntityNameAndEventType(object.getClass().getSimpleName(), "post-update");
            hooks.stream().forEach(wh -> webhookProcessor.notifyWebhook(wh, object));
        }
    });

}

/**
 * <strong>此方法仅限与bdf2-core模块配合使用,且只能用于不加事务的方法当中，如果当前方法或当前方便的调用方法需要用到事务，一定不能使用该方法，否则事务将不起作用</strong><br>
 * 此方法返回当前线程当中指定的数据源对应的Session,如线程中未指定，则返回默认的Session,<br>
 * 注意此方法返回的Session对象我们在使用完毕之后，无需将其close，<br>
 * 在线程执行完毕时，BDF会自动将此方法产生的Session对象进行close处理；同时，此方法仅限在标准的WEB请求调用时使用，<br>
 * 如果在后台的JOB中调用此方法可能会有异常抛出
 * @return 返回一个标准的Hibernate Session对象
 */
public Session getSession(){
    String dsName=sessionFactoryRepository.getDefaultSessionFactoryName();
    dsName=this.getDataSourceName(dsName);
    if(StringUtils.isEmpty(TransactionSynchronizationManager.getCurrentTransactionName())){
        Map<String,Session> map=ContextHolder.getHibernateSessionMap();
        if(map==null){
            throw new RuntimeException("This method can only used in bdf2-core module");
        }
        if(map.containsKey(dsName) && map.get(dsName).isOpen()){
            return map.get(dsName);             
        }else{
            if(map.containsKey(dsName)){
                map.remove(dsName);
            }
            Session session=getSessionFactory().openSession();
            map.put(dsName, session);
            return session;
        }
    }else{
        return this.getSessionFactory(dsName).getCurrentSession();
    }
}

/**
 * Unbind the Hibernate {@code Session} from the thread and close it (in
 * single session mode), or process deferred close for all sessions that have
 * been opened during the current request (in deferred close mode).
 * @see org.springframework.transaction.support.TransactionSynchronizationManager
 */
@Override
public void afterCompletion(WebRequest request, Exception ex) throws DataAccessException {
    if (!decrementParticipateCount(request)) {
        if (isSingleSession()) {
            // single session mode
            SessionHolder sessionHolder =
                    (SessionHolder) TransactionSynchronizationManager.unbindResource(getSessionFactory());
            logger.debug("Closing single Hibernate Session in OpenSessionInViewInterceptor");
            SessionFactoryUtils.closeSession(sessionHolder.getSession());
        }
        else {
            // deferred close mode
            SessionFactoryUtils.processDeferredClose(getSessionFactory());
        }
    }
}

/**
 * Retrieves the primary key from {@code acl_class}, creating a new row if needed and the
 * {@code allowCreate} property is {@code true}.
 *
 * @param type to find or create an entry for (often the fully-qualified class name)
 * @param allowCreate true if creation is permitted if not found
 *
 * @return the primary key or null if not found
 */
protected AclClass createOrRetrieveClassPrimaryKey(String type, boolean allowCreate) {
    List<AclClass> classIds = aclDao.findAclClassList(type);

    if (!classIds.isEmpty()) {
        return classIds.get(0);
    }

    if (allowCreate) {
        AclClass clazz = new AclClass();
        clazz.setClazz(type);

        Assert.isTrue(TransactionSynchronizationManager.isSynchronizationActive(),
                "Transaction must be running");
        return aclDao.createAclClass(clazz);
    }

    return null;
}

/**
 * Retrieves the primary key from {@code acl_class}, creating a new row if needed and the
 * {@code allowCreate} property is {@code true}.
 *
 * @param type to find or create an entry for (often the fully-qualified class name)
 * @param allowCreate true if creation is permitted if not found
 *
 * @return the primary key or null if not found
 */
protected AclClass createOrRetrieveClassPrimaryKey(String type, boolean allowCreate) {
    List<AclClass> classIds = aclDao.findAclClassList(type);

    if (!classIds.isEmpty()) {
        return classIds.get(0);
    }

    if (allowCreate) {
        AclClass clazz = new AclClass();
        clazz.setClazz(type);

        Assert.isTrue(TransactionSynchronizationManager.isSynchronizationActive(),
                "Transaction must be running");
        return aclDao.createAclClass(clazz);
    }

    return null;
}

@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
    SessionFactory sf = getSessionFactory();
    if (!TransactionSynchronizationManager.hasResource(sf)) {
        // New Session to be bound for the current method's scope...
        Session session = openSession();
        try {
            TransactionSynchronizationManager.bindResource(sf, new SessionHolder(session));
            return invocation.proceed();
        }
        finally {
            SessionFactoryUtils.closeSession(session);
            TransactionSynchronizationManager.unbindResource(sf);
        }
    }
    else {
        // Pre-bound Session found -> simply proceed.
        return invocation.proceed();
    }
}

@Override
public void execute(final AssignUsersMessage.Request request) throws Exception {
  request.getUserAssignments().forEach(assignment -> {
    final Subject subject = subjectRepository
        .getSubjectForSubjectModelInProcess(request.getPiId(), assignment.getSmId());
    if (subject != null) {
      subject.setUser(assignment.getUserId());
      subjectRepository.save((SubjectImpl) subject);
      LOG.info("New user for subject: {}", subject);
    }
  });

  final ActorRef sender = getSender();
  TransactionSynchronizationManager
      .registerSynchronization(new TransactionSynchronizationAdapter() {
        @Override
        public void afterCommit() {
          sender.tell(new AssignUsersMessage.Response(), getSelf());
        }
      });

}

private void handleStoreExternalDataCommand(final Object obj) {
  final StoreExternalDataCommand cmd = (StoreExternalDataCommand) obj;
  final Message message = messageRepository.findOne(cmd.getId());
  message.setExternalData(cmd.getData());
  message.setMessageState(MessageState.COMPOSED);
  messageRepository.save((MessageImpl) message);
  final String actorId = getContext().parent().path().name();

  final ActorRef sender = getSender();
  TransactionSynchronizationManager
      .registerSynchronization(new TransactionSynchronizationAdapter() {
        @Override
        public void afterCommit() {
          LOG.info("Updated external data of [{}]", message);
          sender.tell(new ComposedMessageEvent(actorId, message.getId()), getSelf());
          stop();
        }
      });
}

/**
 * Retrieves the primary key from {@code acl_class}, creating a new row if needed and the
 * {@code allowCreate} property is {@code true}.
 *
 * @param type to find or create an entry for (often the fully-qualified class name)
 * @param allowCreate true if creation is permitted if not found
 *
 * @return the primary key or null if not found
 */
protected AclClass createOrRetrieveClassPrimaryKey(String type, boolean allowCreate) {
    List<AclClass> classIds = aclDao.findAclClassList(type);

    if (!classIds.isEmpty()) {
        return classIds.get(0);
    }

    if (allowCreate) {
        AclClass clazz = new AclClass();
        clazz.setClazz(type);

        Assert.isTrue(TransactionSynchronizationManager.isSynchronizationActive(),
                "Transaction must be running");
        return aclDao.createAclClass(clazz);
    }

    return null;
}

@Override
public void afterCompletion(int status) {
    // If we haven't closed the Connection in beforeCompletion,
    // close it now. The holder might have been used for other
    // cleanup in the meantime, for example by a Hibernate Session.
    if (this.holderActive) {
        // The thread-bound ConnectionHolder might not be available anymore,
        // since afterCompletion might get called from a different thread.
        TransactionSynchronizationManager.unbindResourceIfPossible(this.dataSource);
        this.holderActive = false;
        if (this.connectionHolder.hasConnection()) {
            releaseConnection(this.connectionHolder.getConnection(), this.dataSource);
            // Reset the ConnectionHolder: It might remain bound to the thread.
            this.connectionHolder.setConnection(null);
        }
    }
    this.connectionHolder.reset();
}

/**
 * Actually obtain a CCI Connection from the given ConnectionFactory.
 * Same as {@link #getConnection}, but throwing the original ResourceException.
 * <p>Is aware of a corresponding Connection bound to the current thread, for example
 * when using {@link CciLocalTransactionManager}. Will bind a Connection to the thread
 * if transaction synchronization is active (e.g. if in a JTA transaction).
 * <p>Directly accessed by {@link TransactionAwareConnectionFactoryProxy}.
 * @param cf the ConnectionFactory to obtain Connection from
 * @return a CCI Connection from the given ConnectionFactory
 * @throws ResourceException if thrown by CCI API methods
 * @see #doReleaseConnection
 */
public static Connection doGetConnection(ConnectionFactory cf) throws ResourceException {
    Assert.notNull(cf, "No ConnectionFactory specified");

    ConnectionHolder conHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(cf);
    if (conHolder != null) {
        return conHolder.getConnection();
    }

    logger.debug("Opening CCI Connection");
    Connection con = cf.getConnection();

    if (TransactionSynchronizationManager.isSynchronizationActive()) {
        logger.debug("Registering transaction synchronization for CCI Connection");
        conHolder = new ConnectionHolder(con);
        conHolder.setSynchronizedWithTransaction(true);
        TransactionSynchronizationManager.registerSynchronization(new ConnectionSynchronization(conHolder, cf));
        TransactionSynchronizationManager.bindResource(cf, conHolder);
    }

    return con;
}

@Override
protected void flushResource(EntityManagerHolder resourceHolder) {
    EntityManager em = resourceHolder.getEntityManager();
    if (em instanceof EntityManagerProxy) {
        EntityManager target = ((EntityManagerProxy) em).getTargetEntityManager();
        if (TransactionSynchronizationManager.hasResource(target)) {
            // ExtendedEntityManagerSynchronization active after joinTransaction() call:
            // flush synchronization already registered.
            return;
        }
    }
    try {
        em.flush();
    }
    catch (RuntimeException ex) {
        if (this.jpaDialect != null) {
            throw this.jpaDialect.translateExceptionIfPossible(ex);
        }
        else {
            throw convertJpaAccessExceptionIfPossible(ex);
        }
    }
}

@Override
public void preHandle(WebRequest request) throws DataAccessException {
    if (TransactionSynchronizationManager.hasResource(getPersistenceManagerFactory())) {
        // Do not modify the PersistenceManager: just mark the request accordingly.
        String participateAttributeName = getParticipateAttributeName();
        Integer count = (Integer) request.getAttribute(participateAttributeName, WebRequest.SCOPE_REQUEST);
        int newCount = (count != null ? count + 1 : 1);
        request.setAttribute(getParticipateAttributeName(), newCount, WebRequest.SCOPE_REQUEST);
    }
    else {
        logger.debug("Opening JDO PersistenceManager in OpenPersistenceManagerInViewInterceptor");
        PersistenceManager pm =
                PersistenceManagerFactoryUtils.getPersistenceManager(getPersistenceManagerFactory(), true);
        TransactionSynchronizationManager.bindResource(
                getPersistenceManagerFactory(), new PersistenceManagerHolder(pm));
    }
}

/**
 * Retrieves the primary key from {@code acl_class}, creating a new row if needed and the
 * {@code allowCreate} property is {@code true}.
 *
 * @param type to find or create an entry for (often the fully-qualified class name)
 * @param allowCreate true if creation is permitted if not found
 *
 * @return the primary key or null if not found
 */
protected AclClass createOrRetrieveClassPrimaryKey(String type, boolean allowCreate) {
    List<AclClass> classIds = aclDao.findAclClassList(type);

    if (!classIds.isEmpty()) {
        return classIds.get(0);
    }

    if (allowCreate) {
        AclClass clazz = new AclClass();
        clazz.setClazz(type);

        Assert.isTrue(TransactionSynchronizationManager.isSynchronizationActive(),
                "Transaction must be running");
        return aclDao.createAclClass(clazz);
    }

    return null;
}

/**
 * Prepare the given Criteria object, applying cache settings and/or
 * a transaction timeout.
 * @param criteria the Criteria object to prepare
 * @see #setCacheQueries
 * @see #setQueryCacheRegion
 */
protected void prepareCriteria(Criteria criteria) {
    if (isCacheQueries()) {
        criteria.setCacheable(true);
        if (getQueryCacheRegion() != null) {
            criteria.setCacheRegion(getQueryCacheRegion());
        }
    }
    if (getFetchSize() > 0) {
        criteria.setFetchSize(getFetchSize());
    }
    if (getMaxResults() > 0) {
        criteria.setMaxResults(getMaxResults());
    }

    SessionHolder sessionHolder =
            (SessionHolder) TransactionSynchronizationManager.getResource(getSessionFactory());
    if (sessionHolder != null && sessionHolder.hasTimeout()) {
        criteria.setTimeout(sessionHolder.getTimeToLiveInSeconds());
    }
}

/**
 * Prepare the given Query object, applying cache settings and/or
 * a transaction timeout.
 * @param queryObject the Query object to prepare
 * @see #setCacheQueries
 * @see #setQueryCacheRegion
 */
protected void prepareQuery(Query queryObject) {
    if (isCacheQueries()) {
        queryObject.setCacheable(true);
        if (getQueryCacheRegion() != null) {
            queryObject.setCacheRegion(getQueryCacheRegion());
        }
    }
    if (getFetchSize() > 0) {
        queryObject.setFetchSize(getFetchSize());
    }
    if (getMaxResults() > 0) {
        queryObject.setMaxResults(getMaxResults());
    }

    SessionHolder sessionHolder =
            (SessionHolder) TransactionSynchronizationManager.getResource(getSessionFactory());
    if (sessionHolder != null && sessionHolder.hasTimeout()) {
        queryObject.setTimeout(sessionHolder.getTimeToLiveInSeconds());
    }
}

/**
 * get log context
 * @return
 */
public LogProcessContext getLogProcessContext() {
    LogProcessContext logProcessContext = (LogProcessContext) TransactionSynchronizationManager.getResource(LOG_PROCESS_CONTEXT);
    if(logProcessContext == null){
        throw new RuntimeException("please call TransController.startSoftTrans() before executeMethods!");
    }
    return logProcessContext;
}

/**
 * Apply the current transaction timeout, if any, to the given
 * Hibernate Query object.
 * @param query the Hibernate Query object
 * @param sessionFactory Hibernate SessionFactory that the Query was created for
 * (may be {@code null})
 * @see org.hibernate.Query#setTimeout
 */
public static void applyTransactionTimeout(Query query, SessionFactory sessionFactory) {
    Assert.notNull(query, "No Query object specified");
    if (sessionFactory != null) {
        SessionHolder sessionHolder =
                (SessionHolder) TransactionSynchronizationManager.getResource(sessionFactory);
        if (sessionHolder != null && sessionHolder.hasTimeout()) {
            query.setTimeout(sessionHolder.getTimeToLiveInSeconds());
        }
    }
}

/**
 * Get a unique identifier associated with each transaction of each thread.  Null is returned if
 * no transaction is currently active.
 * 
 * @return Returns the transaction ID, or null if no transaction is present
 */
public static String getTransactionId()
{
    /*
     * Go direct to the synchronizations as we don't want to register a resource if one doesn't exist.
     * This method is heavily used, so the simple Map lookup on the ThreadLocal is the fastest.
     */

    TransactionSynchronizationImpl txnSynch =
            (TransactionSynchronizationImpl) TransactionSynchronizationManager.getResource(RESOURCE_KEY_TXN_SYNCH);
    if (txnSynch == null)
    {
        if (TransactionSynchronizationManager.isSynchronizationActive())
        {
            // need to lazily register synchronizations
            return registerSynchronizations().getTransactionId();
        }
        else
        {
            return null;   // not in a transaction
        }
    }
    else
    {
        return txnSynch.getTransactionId();
    }
}

/**
 * Cleans out transaction resources if present
 */
private static void clearSynchronization()
{
    if (TransactionSynchronizationManager.hasResource(RESOURCE_KEY_TXN_SYNCH))
    {
        Object txnSynch = TransactionSynchronizationManager.unbindResource(RESOURCE_KEY_TXN_SYNCH);
        // done
        if (logger.isDebugEnabled())
        {
            logger.debug("Unbound txn synch:" + txnSynch);
        }
    }
}

/**
 * Helper method to rebind the synchronization to the transaction
 *
 * @param txnSynch TransactionSynchronizationImpl
 */
private static void rebindSynchronization(TransactionSynchronizationImpl txnSynch)
{
    TransactionSynchronizationManager.bindResource(RESOURCE_KEY_TXN_SYNCH, txnSynch);
    if (logger.isDebugEnabled())
    {
        logger.debug("Bound (rebind) txn synch: " + txnSynch);
    }
}

@Override
protected void doResume(Object transaction, Object suspendedResources) {
    SuspendedResourcesHolder resourcesHolder = (SuspendedResourcesHolder) suspendedResources;
    if (TransactionSynchronizationManager.hasResource(getSessionFactory())) {
        // From non-transactional code running in active transaction synchronization
        // -> can be safely removed, will be closed on transaction completion.
        TransactionSynchronizationManager.unbindResource(getSessionFactory());
    }
    TransactionSynchronizationManager.bindResource(getSessionFactory(), resourcesHolder.getSessionHolder());
    if (getDataSource() != null) {
        TransactionSynchronizationManager.bindResource(getDataSource(), resourcesHolder.getConnectionHolder());
    }
}

@Override
protected void doCleanupAfterCompletion(Object transaction) {
    long txId = getTransactionId();
    logger.info(" [LITX]结束事务 [TX ID: " + txId + "]");
    super.doCleanupAfterCompletion(transaction);
    TransactionSynchronizationManager.unbindResource(LitxConstant.TRANSACTION_CONTEXT_KEY);
    TransactionSynchronizationManager.unbindResource(LitxConstant.TRANSACTION_ID_KEY);
}

@Override
public Object intercept(Invocation invocation) throws Throwable {
    boolean synchronizationActive = TransactionSynchronizationManager.isSynchronizationActive();
    if (!synchronizationActive) {
        Object[] objects = invocation.getArgs();
        MappedStatement ms = (MappedStatement) objects[0];
        DynamicDataSourceType dynamicDataSourceType;
        if ((dynamicDataSourceType = cacheMap.get(ms.getId())) == null) {
            //读方法
            if (ms.getSqlCommandType().equals(SqlCommandType.SELECT)) {
                //!selectKey 为自增id查询主键(SELECT LAST_INSERT_ID() )方法，使用主库
                if (ms.getId().contains(SelectKeyGenerator.SELECT_KEY_SUFFIX)) {
                    dynamicDataSourceType = DynamicDataSourceType.WRITE;
                } else {
                    BoundSql boundSql = ms.getSqlSource().getBoundSql(objects[1]);
                    String sql = boundSql.getSql().toLowerCase(Locale.CHINA).replaceAll("[\\t\\n\\r]", " ");
                    if (sql.matches(REGEX)) {
                        dynamicDataSourceType = DynamicDataSourceType.WRITE;
                    } else {
                        dynamicDataSourceType = DynamicDataSourceType.READ;
                    }
                }
            } else {
                dynamicDataSourceType = DynamicDataSourceType.WRITE;
            }
            log.warn("设置方法[{}] use [{}] Strategy, SqlCommandType [{}]..", ms.getId(),
                    dynamicDataSourceType.name(), ms.getSqlCommandType().name());
            cacheMap.put(ms.getId(), dynamicDataSourceType);
        }
        DynamicDataSourceHolder.putDataSource(dynamicDataSourceType);
    }

    return invocation.proceed();
}

@Override
public void suspend() {
    if (this.holderActive) {
        TransactionSynchronizationManager.unbindResource(this.sessionFactory);
        // Eagerly disconnect the Session here, to make release mode "on_close" work on JBoss.
        getCurrentSession().disconnect();
    }
}

private void handleStateObjectChangeMessage(final StateObjectChangeMessage.Request request)
    throws Exception {
  final SubjectState subjectState = Optional
      .ofNullable(
          subjectStateRepository.getSubjectStateOfUser(request.getPiId(), request.getUserId()))
      .get();

  sender = getSender();

  final ActorRef bussinessObjectCheckActor = getContext().actorOf(
      springExtension.props("BusinessObjectCheckActor", subjectState.getCurrentState().getSId()),
      UUID.randomUUID().toString());

  // must block thread since transaction is lost when using completable future
  final Future<Object> future = Patterns.ask(bussinessObjectCheckActor, request, Global.TIMEOUT);
  final boolean correct =
      ((Boolean) Await.result(future, Global.TIMEOUT.duration())).booleanValue();

  if (!correct) {
    sender.tell(new Status.Failure(
        new IllegalArgumentException("Check of business objects returned false")), getSelf());
  } else {
    initBusinessObjectInstances(subjectState, request);
    setValuesOfBusinessObjectFieldInstances(subjectState.getCurrentState(), request);
    sendMessages(subjectState, request);

    TransactionSynchronizationManager
        .registerSynchronization(new TransactionSynchronizationAdapter() {
          @Override
          public void afterCommit() {
            sender.tell(new EmptyMessage(), getSelf());
            handleAdditionalActions(subjectState);
          }
        });
  }
}

/**
 * Open a new Hibernate {@code Session} according to the settings of this
 * {@code HibernateAccessor} and bind it to the thread via the
 * {@link TransactionSynchronizationManager}.
 * @see org.springframework.orm.hibernate3.SessionFactoryUtils#getSession
 */
@Override
public void preHandle(WebRequest request) throws DataAccessException {
    String participateAttributeName = getParticipateAttributeName();

    WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);
    if (asyncManager.hasConcurrentResult()) {
        if (applySessionBindingInterceptor(asyncManager, participateAttributeName)) {
            return;
        }
    }

    if ((isSingleSession() && TransactionSynchronizationManager.hasResource(getSessionFactory())) ||
        SessionFactoryUtils.isDeferredCloseActive(getSessionFactory())) {
        // Do not modify the Session: just mark the request accordingly.
        Integer count = (Integer) request.getAttribute(participateAttributeName, WebRequest.SCOPE_REQUEST);
        int newCount = (count != null ? count + 1 : 1);
        request.setAttribute(getParticipateAttributeName(), newCount, WebRequest.SCOPE_REQUEST);
    }
    else {
        if (isSingleSession()) {
            // single session mode
            logger.debug("Opening single Hibernate Session in OpenSessionInViewInterceptor");
            Session session = SessionFactoryUtils.getSession(
                    getSessionFactory(), getEntityInterceptor(), getJdbcExceptionTranslator());
            applyFlushMode(session, false);
            SessionHolder sessionHolder = new SessionHolder(session);
            TransactionSynchronizationManager.bindResource(getSessionFactory(), sessionHolder);

            AsyncRequestInterceptor asyncRequestInterceptor =
                    new AsyncRequestInterceptor(getSessionFactory(), sessionHolder);
            asyncManager.registerCallableInterceptor(participateAttributeName, asyncRequestInterceptor);
            asyncManager.registerDeferredResultInterceptor(participateAttributeName, asyncRequestInterceptor);
        }
        else {
            // deferred close mode
            SessionFactoryUtils.initDeferredClose(getSessionFactory());
        }
    }
}

@After
public void tearDown() {
    try {
        em.close();
    } finally {
        TransactionSynchronizationManager.clear();
    }
}

protected void publishEventAfterCommit(final ApplicationEvent<?> event)
{
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter()
    {
        @Override
        public void afterCommit()
        {
            eventService.publishApplicationEvent(event);
        }
    });
}

protected void publishEventAfterCommit(final ApplicationEvent<?> event)
{
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter()
    {
        @Override
        public void afterCommit()
        {
            eventService.publishApplicationEvent(event);
        }
    });
}

@Override
protected void doResume(Object transaction, Object suspendedResources) {
    SuspendedResourcesHolder resourcesHolder = (SuspendedResourcesHolder) suspendedResources;
    if (TransactionSynchronizationManager.hasResource(getSessionFactory())) {
        // From non-transactional code running in active transaction synchronization
        // -> can be safely removed, will be closed on transaction completion.
        TransactionSynchronizationManager.unbindResource(getSessionFactory());
    }
    TransactionSynchronizationManager.bindResource(getSessionFactory(), resourcesHolder.getSessionHolder());
    if (getDataSource() != null) {
        TransactionSynchronizationManager.bindResource(getDataSource(), resourcesHolder.getConnectionHolder());
    }
}

private void publishEventAfterCommit(final ApplicationEvent<?> event)
{
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter()
    {
        @Override
        public void afterCommit()
        {
            eventService.publishApplicationEvent(event);
        }
    });
}

protected void perItemPostProcess(final ItemOperationParams params)
{
    final List<ApplicationEvent<?>> eventList = new ArrayList<ApplicationEvent<?>>(params.getAfterCommitEvents());
    final List<Runnable> afterCommitHooks = new ArrayList<Runnable>(params.getAfterCommit());
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter()
    {
        @Override
        public void afterCommit()
        {
            afterCommitOne(eventList, afterCommitHooks);
        }
    });
}

private void postProcessParameters(final ItemOperationParams params)
{
    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter()
    {
        @Override
        public void afterCommit()
        {
            afterCommitAll(params);
        }
    });
}