如何使原子成为 ConcurrentHashMaps 上的嵌套迭代操作？

Question

我有一个 ConcurrentHashMap subscriptions 包含另一个对象 (sessionCollection)，我需要执行以下迭代操作:

  subscriptions.values().forEach(sessionCollection -> 
     sessionCollection.removeAllSubscriptionsOfSession(sessionId));

其中 sessionCollection.removeAllSubscriptionsOfSession 对 sessionCollection:

中的集合（也是 ConcurrentHashMap）执行另一个迭代操作

// inside SessionCollection:
private final ConcurrentHashMap<String, CopyOnWriteArrayList<String>> topicsToSessions = 
 new ConcurrentHashMap<>();

public void removeAllSubscriptionsOfSession(String sessionId) {
    // Remove sessions from all topics on record
    topicsToSessions.keySet().forEach(topicSessionKey ->
     removeTopicFromSession(sessionId, topicSessionKey));
}

进行这个整体原子操作的步骤是什么？

Answer 1

ConcurrentHashMap 具有批处理操作 (forEach*())，但它们对于整个地图而言不是原子的。在地图上进行原子批量更改的唯一方法是自己实现所有必要的同步。例如，通过显式使用 synchronized 块或通过为您的地图创建一个包装器（或扩展）来在需要时处理同步。在这种情况下，一个简单的 HashMap 就足够了，因为无论如何您都必须进行同步：

public class SubscriptionsRegistry {
    private final Map<Integer, SessionCollection> map = new HashMap<>();

    public synchronized void removeSubscriptions(Integer sessionId) {
        map.values().forEach(...);
    }

    public synchronized void addSubscription(...) {
        ...
    }

    ...
}

您还需要保护主题到会话的映射（至少是它们的可修改版本）以免泄露到您的 SubscriptionsRegistry 之外，因此没有人能够在没有适当同步的情况下修改它们。

如何使原子成为 ConcurrentHashMaps 上的嵌套迭代操作？

How to make atomic a nested iterative operation on ConcurrentHashMaps?

java

concurrency

multithreading

thread-safety

java.util.concurrent