import com.hazelcast.transaction.TransactionOptions; //导入依赖的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.TransactionOptions; //导入依赖的package包/类
@Override
public LinkResourceAllocations getAllocations(IntentId intentId) {
    checkNotNull(intentId);
    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 {
        STxMap<IntentId, LinkResourceAllocations> intentAllocs = getIntentAllocs(tx);
        return intentAllocs.get(intentId);
    } finally {
        tx.rollbackTransaction();
    }
}
 

import com.hazelcast.transaction.TransactionOptions; //导入依赖的package包/类
@Setup
public void setup() {
    map = targetInstance.getMap(name);
    resultList = targetInstance.getList(name + "results");

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

import com.hazelcast.transaction.TransactionOptions; //导入依赖的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.TransactionOptions; //导入依赖的package包/类
@Override
public StoreTransaction beginTransaction(StoreTxConfig txConfig) throws StorageException {
    if (transactional) {
        TransactionOptions txo = new TransactionOptions();
        txo.setTimeout(lockExpireMS, TimeUnit.MILLISECONDS);
        return new HazelCastTransaction(manager.newTransactionContext(txo), txConfig);
    } else {
        return new NoOpStoreTransaction(txConfig);
    }
}
 

import com.hazelcast.transaction.TransactionOptions; //导入依赖的package包/类
@Override
public <T> T executeTransaction(TransactionOptions transactionOptions, TransactionalTask<T> transactionalTask) throws TransactionException {
    return null;
}
 

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

import com.hazelcast.transaction.TransactionOptions; //导入依赖的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.TransactionOptions; //导入依赖的package包/类
@Override
public <T> T executeTransaction(final TransactionOptions options, final TransactionalTask<T> task) throws TransactionException {
  return getHazelcastInstance().executeTransaction(options, task);
}
 

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

import com.hazelcast.transaction.TransactionOptions; //导入依赖的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.TransactionOptions; //导入依赖的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();

        }
    }