Java 类com.hazelcast.core.TransactionalMap 实例源码

@TimeStep(prob = 0.1)
public void put(ThreadState state) {
    final int key = state.randomKey();
    final int value = state.randomValue();
    targetInstance.executeTransaction(new TransactionalTask<Object>() {
        @Override
        public Object execute(TransactionalTaskContext transactionalTaskContext) {
            TransactionalMap<Integer, Integer> txMap = transactionalTaskContext.getMap(map.getName());
            if (useSet) {
                txMap.set(key, value);
            } else {
                txMap.put(key, value);
            }
            return null;
        }
    });
}

void transfer(int thief, UTS loot) {
  final UTS bag = this.bag.trim();
  final int me = this.me;
  hz.executeTransaction((TransactionalTaskContext context) -> {
    final TransactionalMap<Integer, UTS> map = context.getMap("uts");
    map.set(me, bag);
    final UTS old = map.getForUpdate(thief);
    loot.count = old == null ? 0 : old.count;
    map.set(thief, loot);
    return null;
  });
}

void transfer(int thief, UTS loot) {
  final UTS bag = this.bag.trim();
  final int me = this.me;
  final int wave = ResilientUTS.this.wave;
  hz.executeTransaction((TransactionalTaskContext context) -> {
    final TransactionalMap<Integer, UTS> map = context.getMap("map" + wave);
    map.set(me, bag);
    final UTS old = map.getForUpdate(thief);
    loot.count = old == null ? 0 : old.count;
    map.set(thief, loot);
    return null;
  });
}

@TimeStep(prob = -1)
public void get(ThreadState state) {
    final int key = state.randomKey();
    targetInstance.executeTransaction(new TransactionalTask<Object>() {
        @Override
        public Object execute(TransactionalTaskContext transactionalTaskContext) {
            TransactionalMap<Integer, Integer> txMap = transactionalTaskContext.getMap(map.getName());
            txMap.get(key);
            return null;
        }
    });
}

@TimeStep
public void timeStep(ThreadState state) throws Exception {
    final int key = state.randomInt(keyCount);
    final int increment = state.randomInt(100);

    try {
        targetInstance.executeTransaction(transactionOptions, new TransactionalTask<Object>() {
            @Override
            public Object execute(TransactionalTaskContext txContext) {
                TransactionalMap<Integer, Long> txMap = txContext.getMap(name);
                Long value;
                if (getForUpdate) {
                    value = txMap.getForUpdate(key);
                } else {
                    value = txMap.get(key);
                }
                txMap.put(key, value + increment);
                return null;
            }
        });
        state.increments[key] += increment;
    } catch (TransactionException e) {
        if (reThrowTransactionException) {
            throw rethrow(e);
        }
        logger.warn(name + ": caught TransactionException ", e);
    }
}

@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();
    }
}

/**
 * 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);
        }
    }
}

private STxMap<IntentId, LinkResourceAllocations> getIntentAllocs(TransactionContext tx) {
    TransactionalMap<byte[], byte[]> raw = tx.getMap(INTENT_ALLOCATIONS);
    return new STxMap<>(raw, serializer);
}

private STxMap<LinkKey, List<LinkResourceAllocations>> getLinkAllocs(TransactionContext tx) {
    TransactionalMap<byte[], byte[]> raw = tx.getMap(LINK_RESOURCE_ALLOCATIONS);
    return new STxMap<>(raw, serializer);
}

@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);
}

/**
 * Creates a STxMap instance.
 *
 * @param baseMap base IMap to use
 * @param serializer serializer to use for both key and value
 */
public STxMap(TransactionalMap<byte[], byte[]> baseMap, StoreSerializer serializer) {
    this.m = checkNotNull(baseMap);
    this.serializer = checkNotNull(serializer);
}