多线程：持续写入共享数据结构并定期清除

Question

我有一个问题，我有多个线程不断写入，例如并发 HashMap。现在，我想定期处理（通过 TimerJob）该哈希图中的所有内容。其他线程仍然可以继续写入（这个新数据将在 Timejob 下次启动时处理）。我想知道实现这一目标的最佳方法是什么。我正在阅读，这个问题似乎很像Triple Buffer。尽管如此，我并不是非常积极。有什么想法吗？编辑：我想以这种方式处理后从地图中删除数据，我最终不会重新处理该数据编辑：我不需要将数据写入 HashMap/Set。我只需要将它放在一个集合中，我可以在其他线程仍在写入时定期处理该集合。

Answer 1

我不确定您是否需要地图中的所有数据，或者地图中不再需要的计时器作业处理的数据。

如果您只需要计时器作业的快照之类的东西，您可以像这样用新地图切换/替换地图。

private volatile ConcurentHashMap map ;

public void processByTimerJob(){

   ConcurentHashMap oldMap = this.map;

   this.map = new ConcurrentHashMap; // everyting new will be stored in new map    

   oldMap.forEach(.....   //process old map via iteration or whatever you want

}

Answer 2

我会使用double buffering和读/写锁。

双缓冲通过允许处理换出的映射来减少阻塞。

使用读/写锁让我可以确定在我们交换后没有人仍在写入地图。

class DoubleBufferedMap<K, V> extends AbstractMap<K, V> implements Map<K, V> {

    // Used whenever I want to create a new map.

    private final Supplier<Map<K, V>> mapSupplier;

    // The underlying map.

    private volatile Map<K, V> map;

    // My lock - for ensuring no-one is writing.

    ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

    final Lock readLock = readWriteLock.readLock();

    final Lock writeLock = readWriteLock.writeLock();

    public DoubleBufferedMap(Supplier<Map<K, V>> mapSupplier) {

        this.mapSupplier = mapSupplier;

        this.map = mapSupplier.get();

    }

    /**

     * Swaps out the current map with a new one.

     * 

     * @return the old map ready for processing, guaranteed to have no pending writes.

     */

    public Map<K,V> swap() {

        // Grab the existing map.

        Map<K,V> oldMap = map;

        // Replace it.

        map = mapSupplier.get();

        // Take a write lock to wait for all `put`s to complete.

        try {

            writeLock.lock();

        } finally {

            writeLock.unlock();

        }

        return oldMap;

    }

    // Put methods must take a read lock (yeah I know it's weird)

    @Nullable

    @Override

    public V put(K key, V value) {

        try{

            // Take a read-lock so they know when I'm finished.

            readLock.lock();

            return map.put(key, value);

        } finally {

            readLock.unlock();

        }

    }

    @Override

    public void putAll(@NotNull Map<? extends K, ? extends V> m) {

        try{

            // Take a read-lock so they know when I'm finished.

            readLock.lock();

            map.putAll(m);

        } finally {

            readLock.unlock();

        }

    }

    @Nullable

    @Override

    public V putIfAbsent(K key, V value) {

        try{

            // Take a read-lock so they know when I'm finished.

            readLock.lock();

            return map.putIfAbsent(key, value);

        } finally {

            readLock.unlock();

        }

    }

    // All other methods are simply delegated - but you may wish to disallow some of them (like `entrySet` which would expose the map to modification without locks).

    @Override

    public Set<Entry<K, V>> entrySet() {

        return map.entrySet();

    }

    @Override

    public boolean equals(Object o) {

        return map.equals(o);

    }

    @Override

    public int hashCode() {

        return map.hashCode();

    }

    // ... The rest of the delegators (left to the student)