创建一个支持“快照”的 ConcurrentHashMap

Question

我正在尝试创建一个支持“快照”的ConcurrentHashMap，以提供一致的迭代器，我想知道是否有更有效的方法来做到这一点。问题是，如果同时创建了两个迭代器，那么它们需要读取相同的值，而并发哈希映射的弱一致性迭代器的定义并不能保证会是这种情况。如果可能的话，我还想避免锁定：映射中有几千个值，处理每个项目需要几十毫秒，我不想在这段时间内阻止写入器，因为这可能会导致写入器阻塞一分钟或更长时间。

到目前为止我所拥有的：

1. ConcurrentHashMap's 键是字符串，它的值是 ConcurrentSkipListMap<Long, T> 的实例
2. 当使用putIfAbsent 将元素添加到hashmap 时，会分配一个新的skiplist，并通过skipList.put(System.nanoTime(), t) 添加对象。
3. 为了查询地图，我使用map.get(key).lastEntry().getValue() 返回最新的值。为了查询快照（例如使用迭代器），我使用map.get(key).lowerEntry(iteratorTimestamp).getValue()，其中iteratorTimestamp 是在迭代器初始化时调用System.nanoTime() 的结果。
4. 如果一个对象被删除，我使用map.get(key).put(timestamp, SnapShotMap.DELETED)，其中DELETED是一个静态的最终对象。

问题：

1. 是否有一个库已经实现了这个？或者除此之外，有没有比ConcurrentHashMap 和ConcurrentSkipListMap 更合适的数据结构？我的键是可比较的，所以也许某种并发树比并发哈希表更能支持快照。

2. 如何防止这个东西不断增长？在 X 上或之前初始化的所有迭代器完成之后，我可以删除键小于 X 的所有跳过列表条目（映射中的最后一个键除外），但我不知道确定何时的好方法这已经发生了：当迭代器的 hasNext 方法返回 false 时，我可以标记它已完成，但并非所有迭代器都必须运行完成；我可以将WeakReference 保留到迭代器，以便我可以检测到它何时被垃圾收集，但我想不出一个好的方法来检测这个，除了使用一个遍历弱引用集合的线程，然后休眠几分钟 - 理想情况下，线程会在 WeakReference 上阻塞，并在包装​​的引用被 GC 时收到通知，但我不认为这是一个选项。

   ConcurrentSkipListMap<Long, WeakReference<Iterator>> iteratorMap;
   while(true) {
       long latestGC = 0;
       for(Map.Entry<Long, WeakReference<Iterator>> entry : iteratorMap.entrySet()) {
           if(entry.getValue().get() == null) {
               iteratorMap.remove(entry.getKey());
               latestGC = entry.getKey();
           } else break;
       }
       // remove ConcurrentHashMap entries with timestamps less than `latestGC`
       Thread.sleep(300000); // five minutes
   }
   

编辑：为了澄清答案和 cmets 中的一些混淆，我目前正在将弱一致的迭代器传递给公司另一个部门编写的代码，他们要求我增加强度迭代器的一致性。他们已经意识到我做 100% 一致的迭代器是不可行的，他们只希望我尽最大努力。他们更关心吞吐量而不是迭代器的一致性，因此粗粒度锁不是一种选择。

Answer 1

您需要特殊实现的实际用例是什么？来自ConcurrentHashMap 的Javadoc（强调）：

检索反映了最近完成的更新操作在开始时保持的结果。 ... 迭代器和枚举返回反映哈希表在创建迭代器/枚举时或之后的某个时间点的状态的元素。它们不会抛出 ConcurrentModificationException。但是，迭代器被设计为一次只能由一个线程使用。

所以常规的ConcurrentHashMap.values().iterator() 会给你一个“一致的”迭代器，但只供单个线程一次性使用。如果您需要多次和/或通过多个线程使用相同的“快照”，我建议制作地图副本。

编辑：有了新信息和对“强一致”迭代器的坚持，我提供了这个解决方案。请注意，使用 ReadWriteLock 具有以下含义：

• 写入将被序列化（一次只有一个写入器），因此写入性能可能会受到影响。
• 只要没有正在进行的写入，就允许并发读取，因此读取性能的影响应该是最小的。
• 活动的读取器阻止写入器，但只要需要检索对当前“快照”的引用即可。线程拥有快照后，无论处理快照中的信息需要多长时间，它都不再阻塞写入者。
• 当任何写入处于活动状态时，读取器被阻止；写入完成后，所有读者都可以访问新快照，直到有新的写入替换它。

一致性是通过序列化写入并在每次写入上制作当前值的副本来实现的。持有对“陈旧”快照的引用的读者可以继续使用旧快照而不必担心修改，并且垃圾收集器将在没有人再使用旧快照时回收旧快照。假设读者没有要求从较早的时间点请求快照。

由于快照可能在多个并发线程之间共享，因此快照是只读的，无法修改。此限制也适用于从快照创建的任何Iterator 实例的remove() 方法。

import java.util.*;
import java.util.concurrent.locks.*;

public class StackOverflow16600019 <K, V> {
    private final ReadWriteLock locks = new ReentrantReadWriteLock();
    private final HashMap<K,V> map = new HashMap<>();
    private Collection<V> valueSnapshot = Collections.emptyList();

    public V put(K key, V value) {
        locks.writeLock().lock();
        try {
            V oldValue = map.put(key, value);
            updateSnapshot();
            return oldValue;
        } finally {
            locks.writeLock().unlock();
        }
    }

    public V remove(K key) {
        locks.writeLock().lock();
        try {
            V removed = map.remove(key);
            updateSnapshot();
            return removed;
        } finally {
            locks.writeLock().unlock();
        }
    }

    public Collection<V> values() {
        locks.readLock().lock();
        try {
            return valueSnapshot; // read-only!
        } finally {
            locks.readLock().unlock();
        }
    }

    /** Callers MUST hold the WRITE LOCK. */
    private void updateSnapshot() {
        valueSnapshot = Collections.unmodifiableCollection(
            new ArrayList<V>(map.values())); // copy
    }
}

Answer 2

解决方案 1) 仅在 put 和迭代上进行同步怎么样？这应该会给你一个一致的快照。

Solution2) 开始迭代并生成一个布尔值，然后覆盖 puts、putAll 以便它们进入队列，当迭代完成时，只需使用更改后的值创建这些 puts。

Answer 3

我发现ctrie 是理想的解决方案 - 它是具有恒定时间快照的并发哈希数组映射树