import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
@Override
public Set<ResourceAllocation> getFreeResources(Link link) {
    TransactionOptions opt = new TransactionOptions();
    // read-only and will never be commited, thus does not need durability
    opt.setTransactionType(TransactionType.LOCAL);
    TransactionContext tx = theInstance.newTransactionContext(opt);
    tx.beginTransaction();
    try {
        Map<ResourceType, Set<? extends ResourceAllocation>> freeResources = getFreeResourcesEx(tx, link);
        Set<ResourceAllocation> allFree = new HashSet<>();
        for (Set<? extends ResourceAllocation> r : freeResources.values()) {
            allFree.addAll(r);
        }
        return allFree;
    } finally {
        tx.rollbackTransaction();
    }

}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
private Iterable<LinkResourceAllocations> getAllocations(TransactionContext tx,
                                                         Link link) {
    checkNotNull(tx);
    checkNotNull(link);
    final LinkKey key = LinkKey.linkKey(link);

    STxMap<LinkKey, List<LinkResourceAllocations>> linkAllocs = getLinkAllocs(tx);
    List<LinkResourceAllocations> res = null;
    res = linkAllocs.get(key);
    if (res == null) {
        res = linkAllocs.putIfAbsent(key, new ArrayList<>());
        if (res == null) {
            return Collections.emptyList();
        } else {
            return res;
        }
    }
    return res;
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
@TimeStep
public void timestep(ThreadState state) {
    int key = state.nextRandom(0, range / 2);

    TransactionOptions transactionOptions = new TransactionOptions()
            .setTransactionType(transactionType)
            .setDurability(durability);

    TransactionContext transactionContext = targetInstance.newTransactionContext(transactionOptions);

    transactionContext.beginTransaction();

    TransactionalMap<Object, Object> txMap = transactionContext.getMap("map");

    try {
        Object val = txMap.getForUpdate(key);

        if (val != null) {
            key = state.nextRandom(range / 2, range);
        }

        txMap.put(key, (long) key);

        transactionContext.commitTransaction();
    } catch (Exception e) {
        logger.fatal("----------------------tx exception -------------------------", e);

        if (failOnException) {
            throw rethrow(e);
        }

        transactionContext.rollbackTransaction();
    }
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
@TimeStep
public void timeStep(ThreadState state) {
    firstLock.lock();
    try {
        TransactionContext ctx = targetInstance.newTransactionContext();
        try {
            ctx.beginTransaction();

            TransactionalQueue<Integer> queue = ctx.getQueue(name + 'q');
            queue.offer(1);

            secondLock.lock();
            secondLock.unlock();

            queue.take();

            ctx.commitTransaction();
            state.counter.committed++;
        } catch (Exception txnException) {
            try {
                ctx.rollbackTransaction();
                state.counter.rolled++;

                logger.fatal(name + ": Exception in txn " + state.counter, txnException);
            } catch (Exception rollException) {
                state.counter.failedRollbacks++;
                logger.fatal(name + ": Exception in roll " + state.counter, rollException);
            }
        }
    } catch (Exception e) {
        logger.fatal(name + ": outer Exception" + state.counter, e);
    } finally {
        firstLock.unlock();
    }
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
@Override
public TransactionContext newTransactionContext() {
    return null;
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
/**
 * This method performs transactional operation on removing the {@code exchange}
 * from the operational storage and moving it into the persistent one if the {@link HazelcastAggregationRepository}
 * runs in recoverable mode and {@code optimistic} is false. It will act at <u>your own</u> risk otherwise.
 * @param camelContext   the current CamelContext
 * @param key            the correlation key
 * @param exchange       the exchange to remove
 */
@Override
public void remove(CamelContext camelContext, String key, Exchange exchange) {
    DefaultExchangeHolder holder = DefaultExchangeHolder.marshal(exchange, true, allowSerializedHeaders);
    if (optimistic) {
        LOG.trace("Removing an exchange with ID {} for key {} in an optimistic manner.", exchange.getExchangeId(), key);
        if (!cache.remove(key, holder)) {
            LOG.error("Optimistic locking failed for exchange with key {}: IMap#remove removed no Exchanges, while it's expected to remove one.",
                    key);
            throw new OptimisticLockingException();
        }
        LOG.trace("Removed an exchange with ID {} for key {} in an optimistic manner.", exchange.getExchangeId(), key);
        if (useRecovery) {
            LOG.trace("Putting an exchange with ID {} for key {} into a recoverable storage in an optimistic manner.",
                    exchange.getExchangeId(), key);
            persistedCache.put(exchange.getExchangeId(), holder);
            LOG.trace("Put an exchange with ID {} for key {} into a recoverable storage in an optimistic manner.",
                    exchange.getExchangeId(), key);
        }
    } else {
        if (useRecovery) {
            LOG.trace("Removing an exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(), key);
            // The only considerable case for transaction usage is fault tolerance:
            // the transaction will be rolled back automatically (default timeout is 2 minutes)
            // if no commit occurs during the timeout. So we are still consistent whether local node crashes.
            TransactionOptions tOpts = new TransactionOptions();

            tOpts.setTransactionType(TransactionOptions.TransactionType.LOCAL);
            TransactionContext tCtx = hzInstance.newTransactionContext(tOpts);

            try {
                tCtx.beginTransaction();

                TransactionalMap<String, DefaultExchangeHolder> tCache = tCtx.getMap(cache.getName());
                TransactionalMap<String, DefaultExchangeHolder> tPersistentCache = tCtx.getMap(persistedCache.getName());

                DefaultExchangeHolder removedHolder = tCache.remove(key);
                LOG.trace("Putting an exchange with ID {} for key {} into a recoverable storage in a thread-safe manner.",
                        exchange.getExchangeId(), key);
                tPersistentCache.put(exchange.getExchangeId(), removedHolder);

                tCtx.commitTransaction();
                LOG.trace("Removed an exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(), key);
                LOG.trace("Put an exchange with ID {} for key {} into a recoverable storage in a thread-safe manner.",
                        exchange.getExchangeId(), key);
            } catch (Throwable throwable) {
                tCtx.rollbackTransaction();

                final String msg = String.format("Transaction with ID %s was rolled back for remove operation with a key %s and an Exchange ID %s.",
                        tCtx.getTxnId(), key, exchange.getExchangeId());
                LOG.warn(msg, throwable);
                throw new RuntimeException(msg, throwable);
            }
        } else {
            cache.remove(key);
        }
    }
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
private STxMap<IntentId, LinkResourceAllocations> getIntentAllocs(TransactionContext tx) {
    TransactionalMap<byte[], byte[]> raw = tx.getMap(INTENT_ALLOCATIONS);
    return new STxMap<>(raw, serializer);
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
private STxMap<LinkKey, List<LinkResourceAllocations>> getLinkAllocs(TransactionContext tx) {
    TransactionalMap<byte[], byte[]> raw = tx.getMap(LINK_RESOURCE_ALLOCATIONS);
    return new STxMap<>(raw, serializer);
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
@Override
public TransactionContext newTransactionContext() {
  return getHazelcastInstance().newTransactionContext();
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
@Override
@SuppressWarnings("PMD.PreserveStackTrace")
public void run() {
    TransactionOptions options = new TransactionOptions()
            .setTransactionType(transactionType)
            .setDurability(durability);

    while (!testContext.isStopped()) {
        TransactionContext context = null;

        final int key = random.nextInt(keyCount);
        final long increment = random.nextInt(100);
        try {
            context = targetInstance.newTransactionContext(options);
            context.beginTransaction();

            final TransactionalMap<Integer, Long> map = context.getMap(name);

            Long current = map.getForUpdate(key);
            Long update = current + increment;
            map.put(key, update);

            context.commitTransaction();

            // Do local increments if commit is successful, so there is no needed decrement operation
            localIncrements[key] += increment;
            count.committed++;
        } catch (Exception commitFailedException) {
            if (context != null) {
                try {
                    logger.warn(name + ": commit failed key=" + key + " inc=" + increment, commitFailedException);
                    if (rethrowAllException) {
                        throw rethrow(commitFailedException);
                    }

                    context.rollbackTransaction();
                    count.rolled++;
                } catch (Exception rollBackFailedException) {
                    logger.warn(name + ": rollback failed key=" + key + " inc=" + increment,
                            rollBackFailedException);
                    count.failedRollbacks++;

                    if (rethrowRollBackException) {
                        throw rethrow(rollBackFailedException);
                    }
                }
            }
        }
    }
    targetInstance.getList(name + "res").add(localIncrements);
    targetInstance.getList(name + "report").add(count);
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
public HazelCastTransaction(TransactionContext context, StoreTxConfig txConfig) {
    super(txConfig);

    this.context = context;
    this.context.beginTransaction();
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
public static void releaseTransactionContext(
    TransactionContext transactionContext,
    HazelcastInstance hazelcastInstance) {
  // no op?
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
/**
 * <p>
 * Returns a queue that will be bound to the current transaction if there is
 * an active Hazelcast transaction. If there is no active transaction, null is
 * returned. The {@code synchedLocalTransactionAllowed} can be used to allow
 * transaction synchronization with any active transaction, even if it isn't a
 * Hazelcast transaction. This is useful for synchronizing a Hazelcast
 * transaction to a different PlatformTransactionManager such as a JDBC
 * transaction.
 * </p>
 * <p>
 * WARNING: Hazelcast defines two different interfaces for a transactional
 * queue and a non-transactional queue. This method maps them both to the same
 * interface via generated proxies when needed; however the transactional
 * queue only implements a subset of the normal queue operations. Calling an
 * operation that is not supported on a transactional queue will result in a
 * {@link UnsupportedOperationException}.
 * </p>
 *
 * @param <E> the type of the items in the queue
 * @param name the name of the queue to get
 * @param hazelcastInstance the Hazelcast instance to get the queue from
 * @param synchedLocalTransactionAllowed true to allow a new Hazelcast
 * transaction to be started and synchronized with any existing transaction;
 * false to only return the transactional object if a top-level Hazelcast
 * transaction is active
 *
 * @return the transactional queue if there is an active Hazelcast transaction
 * or an active transaction to synchronize to and
 * synchedLocalTransactionAllowed is true; null otherwise
 */
public static <E> IQueue<E> getTransactionalQueue(String name,
    HazelcastInstance hazelcastInstance,
    boolean synchedLocalTransactionAllowed) {

  TransactionContext transactionContext =
      getHazelcastTransactionContext(hazelcastInstance,
          synchedLocalTransactionAllowed);

  if (transactionContext != null) {
    TransactionalQueue targetQueue = transactionContext.getQueue(name);
    return QueueTopicProxyFactory.createQueueProxy(targetQueue);
  }
  else {
    // No transaction to synchronize to.
    return null;
  }
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
/**
 * Returns the transaction context holder bound to the current transaction. If
 * one cannot be found and a transaction is active, a new one will be created
 * and bound to the thread. If one cannot be found and a transaction is not
 * active, this method returns null.
 *
 * @param hazelcastInstance the Hazelcast instance used to create begin a
 * transaction if needed
 *
 * @return the transaction context holder if a transaction is active, null
 * otherwise
 */
private static TransactionContext getHazelcastTransactionContext(
    HazelcastInstance hazelcastInstance,
    boolean synchedLocalTransactionAllowed) {

  HazelcastTransactionContextHolder conHolder =
      (HazelcastTransactionContextHolder) TransactionSynchronizationManager.
      getResource(hazelcastInstance);

  if (conHolder != null && (conHolder.hasTransactionContext() || conHolder.
      isSynchronizedWithTransaction())) {
    // We are already synchronized with the transaction which means
    // someone already requested a transactional resource or the transaction
    // is being managed at the top level by HazelcastTransactionManager.

    if (!conHolder.hasTransactionContext()) {
      // I think this means we are synchronized with the transaction but
      // we don't have a transactional context because the transaction was
      // suspended. I don't see how we can do this with Hazelcast because
      // it binds the transaction context to the thread and doesn't
      // supported nested transactions. Maybe I'm missing something.

      throw new NestedTransactionNotSupportedException("Trying to resume a "
          + "Hazelcast transaction? Can't do that.");
    }
  }
  else if (TransactionSynchronizationManager.isSynchronizationActive()
      && synchedLocalTransactionAllowed) {
    // No holder or no transaction context but we want to be
    // synchronized to the transaction.
    if (conHolder == null) {
      conHolder = new HazelcastTransactionContextHolder();
      TransactionSynchronizationManager.bindResource(hazelcastInstance,
          conHolder);
    }

    TransactionContext transactionContext = hazelcastInstance.
        newTransactionContext();
    transactionContext.beginTransaction();

    conHolder.setTransactionContext(transactionContext);
    conHolder.setSynchronizedWithTransaction(true);
    conHolder.setTransactionActive(true);
    TransactionSynchronizationManager.registerSynchronization(
        new HazelcastTransactionSynchronization(conHolder, hazelcastInstance));
  }

  return conHolder != null ? conHolder.getTransactionContext() : null;
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
private static void run() {

        Config config = new Config("queueTest");

        QueueConfig queueConfig = config.getQueueConfig(QNAME);

        QueueStoreConfig queueStoreConfig = new QueueStoreConfig();
        queueStoreConfig.setEnabled(true);
        queueStoreConfig.setStoreImplementation(new MockQueueStore());
        queueStoreConfig.getProperties().setProperty("memory-limit", "0");

        queueConfig.setQueueStoreConfig(queueStoreConfig);

        HazelcastInstance hzInstance = Hazelcast.newHazelcastInstance(config);

        long startTime = System.currentTimeMillis();

        int i = 0;
        while (i++ < 2000000) {
            if (i % 10000 == 0) {
                logger.info(Integer.toString(i) + "\t" + String.format("%8.3f", (double) (System.currentTimeMillis() -
                        startTime) / i));
            }

            TransactionOptions options = new TransactionOptions().setTransactionType(TransactionOptions.TransactionType.LOCAL);

            TransactionContext context = hzInstance.newTransactionContext(options);
            context.beginTransaction();

            TransactionalQueue<Integer> queue = context.getQueue(QNAME);
            queue.offer(i);

            context.commitTransaction();

        }
    }
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
/**
 * Constructs the context holder.
 *
 * @param transactionContext the transaction context to hold
 */
HazelcastTransactionContextHolder(TransactionContext transactionContext) {
  this.transactionContext = transactionContext;
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
/**
 * Sets the transaction context to be held in the resource holder.
 *
 * @param transactionContext the transaction context
 */
void setTransactionContext(TransactionContext transactionContext) {
  this.transactionContext = transactionContext;
}
 

import com.hazelcast.transaction.TransactionContext; //导入依赖的package包/类
/**
 * Returns the transaction context in the holder or null if none has been set.
 *
 * @return the transaction context or null
 */
TransactionContext getTransactionContext() {
  return this.transactionContext;
}