com.google.common.util.concurrent.AtomicLongMap#create ( )源码实例Demo

private List<Predicate> getSortedPredicates() {

		long[] counts = getPredicateCounts();

		AtomicLongMap<Predicate> pCounts = AtomicLongMap.create();
		for (int i = 0; i < counts.length; ++i) {
			pCounts.put(PredicateSet.getPredicate(i), counts[i]);
		}

		List<Predicate> pOrder = new ArrayList<>(addablePredicates.size());
		for (Predicate p : addablePredicates) {
			pOrder.add(p);
		}

		pOrder.sort(new Comparator<Predicate>() {
			@Override
			public int compare(Predicate o1, Predicate o2) {
				return Long.compare(pCounts.get(o2), pCounts.get(o1));
			}
		});

		return pOrder;
	}
 

public static EurostagFakeNodes build(Network network, EurostagEchExportConfig config) {
    Objects.requireNonNull(network);
    Objects.requireNonNull(config);

    BiMap<String, String> fakeNodesMap = HashBiMap.create(new HashMap<>());
    AtomicLongMap<String> countUsesMap = AtomicLongMap.create();

    //adds 2 default fake nodes
    fakeNodesMap.put(EchUtil.FAKE_NODE_NAME1, EchUtil.FAKE_NODE_NAME1);
    countUsesMap.getAndIncrement(EchUtil.FAKE_NODE_NAME1);
    fakeNodesMap.put(EchUtil.FAKE_NODE_NAME2, EchUtil.FAKE_NODE_NAME2);
    countUsesMap.getAndIncrement(EchUtil.FAKE_NODE_NAME2);

    Identifiables.sort(network.getVoltageLevels()).stream().map(VoltageLevel::getId).forEach(vlId ->
            fakeNodesMap.put(vlId, newEsgId(fakeNodesMap, vlId)));

    return new EurostagFakeNodes(fakeNodesMap, countUsesMap, network);
}
 

@Test
public void testIncrement() throws Exception {
    AtomicLongMap<String> cache = AtomicLongMap.create();
    cache.addAndGet("a", 1L);
    cache.addAndGet("a", 2L);
    cache.addAndGet("b", 5L);


    Map<String, Long> remove = AtomicLongMapUtils.remove(cache);
    Assert.assertEquals((long) remove.get("a"), 3L);
    Assert.assertEquals((long) remove.get("b"), 5L);

    cache.addAndGet("a", 1L);
    Map<String, Long> remove2 = AtomicLongMapUtils.remove(cache);
    Assert.assertEquals((long) remove2.get("a"), 1L);
}
 

private void dumpAllState(DumpData dumpData) {
    AtomicLongMap<Integer> allStateMap = AtomicLongMap.create();
    List<Map<Integer, Long>> allRecord = ImmutableList.of(dumpData.getBlockedTimes(), dumpData.getRunnableCpuTimes(),
            dumpData.getTimedWaitingTimes(), dumpData.getWaitingTimes());
    for (Map<Integer, Long> cpuTime : allRecord) {
        for (Map.Entry<Integer, Long> entry : cpuTime.entrySet()) {
            allStateMap.addAndGet(entry.getKey(), entry.getValue());
        }
    }
    doDump(allStateMap.asMap(), Manager.getAllStatePath(), false);
    doDump(allStateMap.asMap(), Manager.getFilterAllStatePath(), true);
}
 

public ItemFastSlowRateLimiter(Duration fastDelay, Duration slowDelay, int maxFastAttempts) {
  this.fastDelay = fastDelay;
  this.slowDelay = slowDelay;
  this.maxFastAttempts = maxFastAttempts;

  failures = AtomicLongMap.create();
}
 

private AtomicLongMap<PartitionRefiner> createSelectivityEstimation(IEvidenceSet sampleEvidence,
		Set<PredicatePair> predicatePairs) {
	AtomicLongMap<PartitionRefiner> selectivityCount = AtomicLongMap.create();
	for (PredicateBitSet ps : sampleEvidence) {
		int count = (int) sampleEvidence.getCount(ps);
		ps.forEach(p -> {
			selectivityCount.addAndGet(p, count);
		});
		for (PredicatePair pair : predicatePairs)
			if (pair.bothContainedIn(ps)) {
				selectivityCount.addAndGet(pair, sampleEvidence.getCount(ps));
			}
	}
	return selectivityCount;
}
 

public TermVectorsFilter(Fields termVectorsByField, Fields topLevelFields, Set<String> selectedFields, @Nullable AggregatedDfs dfs) {
    this.fields = termVectorsByField;
    this.topLevelFields = topLevelFields;
    this.selectedFields = selectedFields;

    this.dfs = dfs;
    this.scoreTerms = new HashMap<>();
    this.sizes = AtomicLongMap.create();
    this.similarity = new DefaultSimilarity();
}
 

private void tick0() {
  final Instant t0 = time.get();

  final Map<String, Resource> resources;
  final Optional<Long> globalConcurrency;
  final StyxConfig config;
  try {
    config = storage.config();
    globalConcurrency = config.globalConcurrency();
    resources = storage.resources().stream().collect(toMap(Resource::id, identity()));
  } catch (IOException e) {
    log.warn("Failed to read from storage", e);
    return;
  }

  globalConcurrency.ifPresent(
      concurrency ->
          resources.put(GLOBAL_RESOURCE_ID,
              Resource.create(GLOBAL_RESOURCE_ID, concurrency)));

  var activeInstances = stateManager.listActiveInstances();
  var workflows = new ConcurrentHashMap<WorkflowId, Optional<Workflow>>();

  // Note: not a strongly consistent number, so the graphed value can be imprecise or show
  // exceeded limit even if the real usage never exceeded the limit.
  var currentResourceUsage = AtomicLongMap.<String>create();
  var currentResourceDemand = AtomicLongMap.<String>create();

  processInstances(config, resources, workflows, activeInstances, currentResourceUsage, currentResourceDemand);

  // TODO: stats might be inaccurate if some instances fail processing
  updateResourceStats(resources, currentResourceUsage);
  currentResourceDemand.asMap().forEach(stats::recordResourceDemanded);

  final long durationMillis = t0.until(time.get(), ChronoUnit.MILLIS);
  stats.recordTickDuration(TICK_TYPE, durationMillis);

  tracer.getCurrentSpan().addAnnotation("processed",
      Map.of("instances", AttributeValue.longAttributeValue(activeInstances.size())));
}
 

@Test
public void testIntegerMax() throws Exception {
    AtomicLongMap<String> cache = AtomicLongMap.create();
    cache.addAndGet("a", 1L);
    cache.addAndGet("a", 2L);
    cache.addAndGet("b", 5L);
}
 

@Test
public void testIntegerMin() throws Exception {
    AtomicLongMap<String> cache = AtomicLongMap.create();
    cache.addAndGet("a", 1L);
    cache.addAndGet("a", 2L);
    cache.addAndGet("b", 5L);

}
 

/**
 * 以Guava的AtomicLongMap，实现线程安全的HashMap<E,AtomicLong>结构的Counter
 */
public static <E> AtomicLongMap<E> createConcurrentCounterMap() {
	return AtomicLongMap.create();
}
 

/**
 * 以Guava的AtomicLongMap，实现线程安全的HashMap<E,AtomicLong>结构的Counter
 */
public static <E> AtomicLongMap<E> createConcurrentCounterMap() {
	return AtomicLongMap.create();
}
 

public ItemExponentialFailureRateLimiter(Duration baseDelay, Duration maxDelay) {
  this.baseDelay = baseDelay;
  this.maxDelay = maxDelay;

  failures = AtomicLongMap.create();
}
 

/**
 * 以Guava的AtomicLongMap，实现线程安全的HashMap<E,AtomicLong>结构的Counter
 */
public static <E> AtomicLongMap<E> createConcurrentMapCounter() {
	return AtomicLongMap.create();
}
 

public void testRemove_thread_safety() throws InterruptedException {
    final AtomicLongMap<String> cache = AtomicLongMap.create();

    final int totalThread = 5;
    final ExecutorService executorService = Executors.newFixedThreadPool(totalThread);

    final AtomicLong totalCounter = new AtomicLong();
    final AtomicBoolean writerThread = new AtomicBoolean(true);
    final AtomicBoolean removeThread = new AtomicBoolean(true);

    final CountDownLatch writerLatch = new CountDownLatch(totalThread);

    for (int i = 0; i < totalThread; i++) {
        final int writerName = i;
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                while (writerThread.get()) {
                    cache.incrementAndGet("aa");
                    cache.incrementAndGet("cc");
                    cache.incrementAndGet("aa");
                    cache.incrementAndGet("bb");
                    cache.incrementAndGet("bb");
                    cache.incrementAndGet("bb");
                    cache.incrementAndGet("cc");
                    cache.incrementAndGet("d");
                    totalCounter.addAndGet(8);
                }
                writerLatch.countDown();
                logger.debug("shutdown {}", writerName);
            }
        });
    }

    final AtomicLong sumCounter = new AtomicLong();
    executorService.execute(new Runnable() {
        @Override
        public void run() {
            while (removeThread.get()) {
                Map<String, Long> remove = AtomicLongMapUtils.remove(cache);
                sumCounter.addAndGet(sum(remove));
                logger.debug("sum:{}", remove);

                Uninterruptibles.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);
            }
        }
    });

    Uninterruptibles.sleepUninterruptibly(5000, TimeUnit.MILLISECONDS);
    writerThread.set(false);
    writerLatch.await();


    Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
    removeThread.set(false);
    Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);


    executorService.shutdown();
    logger.debug("total={} sum:{}", totalCounter.get(), sumCounter.get());
    Assert.assertEquals("concurrent remove and increment", totalCounter.get(), sumCounter.get());


}
 

private TestAtomicCounterMap(String name) {
    // Init name, map using create
    atomicCounterMapName = name;
    map = AtomicLongMap.create();
}
 

public AtomicLongMapTutorials() {
    atomicLongMap = AtomicLongMap.create();
}