java.util.concurrent.BrokenBarrierException#java.util.concurrent.ConcurrentMap源码实例Demo

protected List<String> getTagEntityList(String host, String tagId) {
    List<String> list = null;
    Map<String, Object> categoryMap = ServiceLocator.getInstance().getMemoryImage("categoryMap");
    ConcurrentMap<Object, Object> listCache = (ConcurrentMap<Object, Object>)categoryMap.get("listCache");
    if(listCache == null) {
        listCache = new ConcurrentLinkedHashMap.Builder<Object, Object>()
                .maximumWeightedCapacity(200)
                .build();
        categoryMap.put("listCache", listCache);
    } else {
        list = (List<String>)listCache.get(host + tagId);
    }
    if(list == null) {
        list = getTagEntityListDb(host, tagId);
        if(list != null) {
            listCache.put(host + tagId, list);
        }
    }
    return list;
}
 

/**
 * Simulates races by modifying the map within the mapping function.
 */
@Test
public void testConcurrentMap_computeIfAbsent_racy() {
    final ConcurrentMap<Long,Long> map = new ImplementsConcurrentMap<>();
    final Long two = 2L;
    Function<Long,Long> f, g;

    // race not detected if function returns null
    f = (k) -> { map.put(two, 42L); return null; };
    assertNull(Maps.computeIfAbsent(map, two, f));
    assertEquals(42L, (long)map.get(two));

    map.clear();
    f = (k) -> { map.put(two, 42L); return 86L; };
    assertEquals(42L, (long) Maps.computeIfAbsent(map, two, f));
    assertEquals(42L, (long) map.get(two));

    // mapping function ignored if value already exists
    map.put(two, 99L);
    assertEquals(99L, (long) Maps.computeIfAbsent(map, two, f));
    assertEquals(99L, (long) map.get(two));
}
 

protected List<String> getRecentEntityList(String host, String categoryType, String sortedBy, String sortDir) {
    List<String> list = null;
    // get list from cache
    Map<String, Object> categoryMap = ServiceLocator.getInstance().getMemoryImage("categoryMap");
    ConcurrentMap<Object, Object> listCache = (ConcurrentMap<Object, Object>)categoryMap.get("listCache");
    if(listCache == null) {
        listCache = new ConcurrentLinkedHashMap.Builder<Object, Object>()
                .maximumWeightedCapacity(200)
                .build();
        categoryMap.put("listCache", listCache);
    } else {
        list = (List<String>)listCache.get(host + categoryType);
    }
    if(list == null) {
        // not in cache, get from db
        list = getRecentEntityListDb(host, categoryType, sortedBy, sortDir);
        if(list != null) {
            listCache.put(host + categoryType, list);
        }
    }
    return list;
}
 

@Override
public CompletableFuture<MeterStoreResult> storeMeter(NetworkId networkId, Meter meter) {

    ConcurrentMap<MeterKey, MeterData> meters = getMetersByNetwork(networkId);

    ConcurrentMap<MeterKey, CompletableFuture<MeterStoreResult>> futures =
           getFuturesByNetwork(networkId);

    CompletableFuture<MeterStoreResult> future = new CompletableFuture<>();
    MeterKey key = MeterKey.key(meter.deviceId(), meter.id());
    futures.put(key, future);
    MeterData data = new MeterData(meter, null, local);

    try {
            meters.put(key, data);
    } catch (StorageException e) {
            future.completeExceptionally(e);
    }

    return future;
}
 

private void sendMyRoutes(String origin) throws IOException {
    PostOffice po = PostOffice.getInstance();
    String myOrigin = Platform.getInstance().getOrigin();
    for (String r : routes.keySet()) {
        ConcurrentMap<String, String> originMap = routes.get(r);
        if (originMap.containsKey(myOrigin)) {
            String personality = originMap.get(myOrigin);
            EventEnvelope request = new EventEnvelope();
            request.setTo(ServiceDiscovery.SERVICE_REGISTRY + "@" + origin)
                    .setHeader(TYPE, ADD)
                    .setHeader(ORIGIN, myOrigin)
                    .setHeader(ROUTE, r).setHeader(PERSONALITY, personality);
            po.send(request);
        }
    }
}
 

/**
 * @param internal Internal entry flag (internal key or not user cache).
 * @param preload Whether update happened during preloading.
 * @return Registered listeners.
 */
@Nullable public Map<UUID, CacheContinuousQueryListener> updateListeners(
    boolean internal,
    boolean preload) {
    if (preload && !internal)
        return null;

    ConcurrentMap<UUID, CacheContinuousQueryListener> lsnrCol;

    if (internal)
        lsnrCol = intLsnrCnt.get() > 0 ? intLsnrs : null;
    else
        lsnrCol = lsnrCnt.get() > 0 ? lsnrs : null;

    return F.isEmpty(lsnrCol) ? null : lsnrCol;
}
 

public List<String> findAddressesByApplication(String application) {
    List<String> ret = new ArrayList<String>();
    ConcurrentMap<String, Map<Long, URL>> providerUrls = getRegistryCache().get(Constants.PROVIDERS_CATEGORY);
    for(Map.Entry<String, Map<Long, URL>> e1 : providerUrls.entrySet()) {
        Map<Long, URL> value = e1.getValue();
        for(Map.Entry<Long, URL> e2 : value.entrySet()) {
            URL u = e2.getValue();
            if(application.equals(u.getParameter(Constants.APPLICATION_KEY))) {
                String addr = u.getAddress();
                if(addr != null) ret.add(addr);
            }
        }
    }
    
    return ret;
}
 

private static Class getDefaultImpl(Class clazz) {
	if (clazz == Collection.class || clazz == List.class) {
		return ArrayList.class;
	}
	if (clazz == Set.class) {
		return HashSet.class;
	}
	if (clazz == Queue.class) {
		return LinkedList.class;
	}
	if (clazz == Deque.class) {
		return ArrayDeque.class;
	}
	if (clazz == Map.class) {
		return HashMap.class;
	}
	if (clazz == ConcurrentMap.class) {
		return ConcurrentHashMap.class;
	}
	if (clazz.isInterface() || Modifier.isAbstract(clazz.getModifiers()))
		throw new RuntimeException("cannot new instance of " + clazz);
	return clazz;
}
 

/**
 * @param ig0 Ignite.
 * @param fullBlt Initial BLT list.
 * @param newBaselineSize New baseline size.
 * @param threadProgressTracker Thread progress tracker.
 */
private void tryChangeBaselineDown(
    IgniteEx ig0,
    List<BaselineNode> fullBlt,
    int newBaselineSize,
    AtomicReference<Throwable> loadError,
    ConcurrentMap<Long, Long> threadProgressTracker
) throws Exception {
    System.out.println("### Changing BLT: " + (newBaselineSize + 1) + " -> " + newBaselineSize);
    ig0.cluster().setBaselineTopology(fullBlt.subList(0, newBaselineSize));

    System.out.println("### Starting rebalancing after BLT change: " + (newBaselineSize + 1) + " -> " + newBaselineSize);
    awaitPartitionMapExchange();
    System.out.println("### Rebalancing is finished after BLT change: " + (newBaselineSize + 1) + " -> " + newBaselineSize);

    awaitProgressInAllLoaders(10_000, loadError, threadProgressTracker);

    if (loadError.get() != null) {
        loadError.get().printStackTrace();

        fail("Unexpected error in load thread: " + loadError.get().toString());
    }
}
 

public XulCacheModel recycle(ConcurrentMap<String, XulCacheModel> caches) {
    XulCacheModel cache = null;
    for (XulRecycleStrategy strategy : _strategies) {
        cache = strategy.findRecycledCache(caches.values());
        if (cache != null) {
            caches.remove(cache.getKey());
            break;
        }
    }

    return cache;
}
 

private void add(ConcurrentMap<String, Collection<Class<?>>> map,
		String pattern, Class<?> role) {
	Collection<Class<?>> list = map.get(pattern);
	if (list == null) {
		list = new CopyOnWriteArrayList<Class<?>>();
		Collection<Class<?>> o = map.putIfAbsent(pattern, list);
		if (o != null) {
			list = o;
		}
	}
	if (!list.contains(role)) {
		list.add(role);
	}
}
 

/**
 * get returns the correct element at the given key,
 * or null if not present
 */
public void testGet() {
    ConcurrentMap map = map5();
    assertEquals("A", (String)map.get(one));
    ConcurrentMap empty = map();
    assertNull(map.get("anything"));
    assertNull(empty.get("anything"));
}
 

@Override
public Map<String, String> getContainerReport() {
  Map<String, String> id2StatMap = new HashMap<>();
  synchronized (containerReportCache) {
    ConcurrentMap<String, ContainerStat> map = containerReportCache.asMap();
    for (Map.Entry<String, ContainerStat> entry : map.entrySet()) {
      id2StatMap.put(entry.getKey(), entry.getValue().toJsonString());
    }
  }

  return id2StatMap;
}
 

/**
 * Returns the connection count in this registry.
 *
 * Note it might count connections that are closed but not removed from registry yet
 *
 * @return the connection count
 */
public int count() {
    int n = 0;
    for (ConcurrentMap<?, ?> bag : registry.values()) {
        n += bag.size();
    }
    return n;
}
 

private static <K,V>ConcurrentMap<K,V> newConcurrentReferenceMap(int initialCapacity){
    if(CONCURRENT_REFERENCE_MAP_CONSTRUCTOR != null){
        try {
            return CONCURRENT_REFERENCE_MAP_CONSTRUCTOR.newInstance(initialCapacity);
        } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
            //skip
        }
    }
    return new ConcurrentHashMap<>(initialCapacity);
}
 

public void removeChildListener(String path, ChildListener listener) {
	ConcurrentMap<ChildListener, TargetChildListener> listeners = childListeners.get(path);
	if (listeners != null) {
		TargetChildListener targetListener = listeners.remove(listener);
		if (targetListener != null) {
			removeTargetChildListener(path, targetListener);
		}
	}
}
 

/**
 * 根据服务id获取服务元信息。
 * @param aliasType 名字分类。
 * @param aliasName 别名。
 */
public RsfBindInfo<?> getRsfBindInfo(String aliasType, String aliasName) {
    ConcurrentMap<String, String> aliasNameMaps = this.aliasNameMap.get(aliasType);
    if (aliasNameMaps == null) {
        return null;
    }
    String serviceID = aliasNameMaps.get(aliasName);
    if (serviceID == null) {
        return null;
    }
    return this.serviceMap.get(serviceID);
}
 

/**
 * Creates a Map of the Appenders.
 * @param appenders An array of Appenders.
 * @return The Appender Map.
 */
@PluginFactory
public static ConcurrentMap<String, Appender> createAppenders(
            @PluginElement("Appenders") final Appender[] appenders) {

    final ConcurrentMap<String, Appender> map =  new ConcurrentHashMap<>(appenders.length);

    for (final Appender appender : appenders) {
        map.put(appender.getName(), appender);
    }

    return map;
}
 

public ConcurrentMap<ExecutionTrace, Integer> getEquivalenceClassMap() {
	final ConcurrentMap<ExecutionTrace, Integer> map = new ConcurrentHashMap<>();
	for (final Entry<AbstractExecutionTraceHashContainer, AtomicInteger> entry : this.eTracesEquivClassesMap.entrySet()) {
		map.put(entry.getKey().getExecutionTrace(), entry.getValue().intValue());
	}
	return map;
}
 

public long queryOffset(final String group, final String topic, final int queueId) {
    // [email protected]
    String key = topic + TOPIC_GROUP_SEPARATOR + group;
    ConcurrentMap<Integer, Long> map = this.offsetTable.get(key);
    if (null != map) {
        Long offset = map.get(queueId);
        if (offset != null)
            return offset;
    }

    return -1;
}
 

private <T> T handleDuplicatedEni(T result, boolean storeOnly) {
    if (!(result instanceof Task)) {
        return result;
    }

    Task task = (Task) result;
    if (task.getTwoLevelResources().isEmpty()) {
        return result;
    }
    TwoLevelResource original = task.getTwoLevelResources().get(0);

    synchronized (twoLevelResourceAssignments) {
        // Store current assignment
        ConcurrentMap<Integer, TwoLevelResource> agentAssignments = twoLevelResourceAssignments.computeIfAbsent(
                task.getTaskContext().getOrDefault(TaskAttributes.TASK_ATTRIBUTES_AGENT_HOST, "DEFAULT"),
                k -> new ConcurrentHashMap<>()
        );
        agentAssignments.put(original.getIndex(), original);

        if (storeOnly) {
            return result;
        }

        // Find another assignment on the same agent with different resource value
        Optional<Task> taskOverride = agentAssignments.values().stream().filter(a -> !a.getValue().equals(original.getValue()))
                .findFirst()
                .map(match -> {
                    TwoLevelResource override = original.toBuilder().withIndex(match.getIndex()).build();
                    return task.toBuilder().withTwoLevelResources(override).build();
                });
        return (T) taskOverride.orElse(task);
    }
}
 

@Override
public void groupOperationFailed(DeviceId deviceId, GroupOperation operation) {

    StoredGroupEntry existing = (groupEntriesById.get(
            deviceId) != null) ?
            groupEntriesById.get(deviceId).get(operation.groupId()) :
            null;

    if (existing == null) {
        log.warn("No group entry with ID {} found ", operation.groupId());
        return;
    }

    switch (operation.opType()) {
    case ADD:
        notifyDelegate(new GroupEvent(Type.GROUP_ADD_FAILED, existing));
        break;
    case MODIFY:
        notifyDelegate(new GroupEvent(Type.GROUP_UPDATE_FAILED, existing));
        break;
    case DELETE:
        notifyDelegate(new GroupEvent(Type.GROUP_REMOVE_FAILED, existing));
        break;
    default:
        log.warn("Unknown group operation type {}", operation.opType());
    }

    ConcurrentMap<GroupKey, StoredGroupEntry> keyTable =
            getGroupKeyTable(existing.deviceId());
    ConcurrentMap<GroupId, StoredGroupEntry> idTable =
            getGroupIdTable(existing.deviceId());
    idTable.remove(existing.id());
    keyTable.remove(existing.appCookie());
}
 

/**
 * keySet returns a Set containing all the keys
 */
public void testKeySet() {
    ConcurrentMap map = map5();
    Set s = map.keySet();
    assertEquals(5, s.size());
    assertTrue(s.contains(one));
    assertTrue(s.contains(two));
    assertTrue(s.contains(three));
    assertTrue(s.contains(four));
    assertTrue(s.contains(five));
}
 

private void putConsumeQueue(final String topic, final int queueId, final ConsumeQueue consumeQueue) {
    ConcurrentMap<Integer/* queueId */, ConsumeQueue> map = this.consumeQueueTable.get(topic);
    if (null == map) {
        map = new ConcurrentHashMap<Integer/* queueId */, ConsumeQueue>();
        map.put(queueId, consumeQueue);
        this.consumeQueueTable.put(topic, map);
    } else {
        map.put(queueId, consumeQueue);
    }
}
 

public HashSet<String> queryTopicConsumeByWho(final String topic) {
        HashSet<String> groups = new HashSet<>();
//        遍历换粗中消费组信息
        Iterator<Entry<String, ConsumerGroupInfo>> it = this.consumerTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, ConsumerGroupInfo> entry = it.next();
//          获取组中的缓存订阅信息
            ConcurrentMap<String, SubscriptionData> subscriptionTable =
                entry.getValue().getSubscriptionTable();
            if (subscriptionTable.containsKey(topic)) {
                groups.add(entry.getKey());
            }
        }
        return groups;
    }
 

private Object[] safeSend(Object[] arguments, ConcurrentMap<QName, URL> overridenEndpointAddresses) throws Exception {
    Object[] results = null;

    results = this.service.getClient().send(this.name, arguments, overridenEndpointAddresses);

    if (results == null) {
        results = new Object[] {};
    }
    return results;
}
 

@Override
public void run(Set<Stats> statsSet) {
    final List<PerfPrinter> perfPrinters = ExtensionLoader.getExtensionLoader(PerfPrinter.class).getExtensions();
    String timestamp = DateTime.now().toString("yyyyMMddHHmmss");
    // 记录当前时间窗口的error数和count值
    ConcurrentMap<String, ConcurrentMap<List<String>, Double>> currentErrCollector = new ConcurrentHashMap<>();
    ConcurrentMap<String, ConcurrentMap<List<String>, Long>> currentSummaryCollector = new ConcurrentHashMap<>();

    for (PerfPrinter perfPrinter : perfPrinters) {
        perfPrinter.print(statsSet, timestamp, currentErrCollector, currentSummaryCollector);
    }

    LastTimeStatsHolder.lastTimeErrCollector = currentErrCollector;
    LastTimeStatsHolder.lastTimeSummaryCollector = currentSummaryCollector;
}
 

/**
 * Return the current set of children at the given path, mapped by child name. There are no
 * guarantees of accuracy; this is merely the most recent view of the data.  If there is no
 * node at this path, {@code null} is returned.
 *
 * @param fullPath full path to the node to check
 * @return a possibly-empty list of children if the node is alive, or null
 */
public Map<String, ChildData> getCurrentChildren(String fullPath)
{
    TreeNode node = find(fullPath);
    if ( node == null || node.nodeState != NodeState.LIVE )
    {
        return null;
    }
    ConcurrentMap<String, TreeNode> map = node.children;
    Map<String, ChildData> result;
    if ( map == null )
    {
        result = ImmutableMap.of();
    }
    else
    {
        ImmutableMap.Builder<String, ChildData> builder = ImmutableMap.builder();
        for ( Map.Entry<String, TreeNode> entry : map.entrySet() )
        {
            TreeNode childNode = entry.getValue();
            ChildData childData = childNode.childData;
            // Double-check liveness after retreiving data.
            if ( childData != null && childNode.nodeState == NodeState.LIVE )
            {
                builder.put(entry.getKey(), childData);
            }
        }
        result = builder.build();
    }

    // Double-check liveness after retreiving children.
    return node.nodeState == NodeState.LIVE ? result : null;
}
 

protected void doUnsubscribe(URL url, NotifyListener listener) {
    ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
    if (listeners != null) {
        ChildListener zkListener = listeners.get(listener);
        if (zkListener != null) {
            zkClient.removeChildListener(toUrlPath(url), zkListener);
        }
    }
}
 

/**
 * @param url url
 */
public static void removeStatus(URL url, String methodName) {
    String uri = url.toIdentityString();
    ConcurrentMap<String, RpcStatus> map = METHOD_STATISTICS.get(uri);
    if (map != null) {
        map.remove(methodName);
    }
}