java 删除条目时迭代 ConcurrentHashMap

Iterate over ConcurrentHashMap while deleting entries

提问by shmosel

I want to periodically iterate over a ConcurrentHashMapwhile removing entries, like this:

我想定期迭代一段ConcurrentHashMap时间删除条目，如下所示：

for (Iterator<Entry<Integer, Integer>> iter = map.entrySet().iterator(); iter.hasNext(); ) {
    Entry<Integer, Integer> entry = iter.next();
    // do something
    iter.remove();
}

The problem is that another thread may be updating or modifying values while I'm iterating. If that happens, those updates can be lost forever, because my thread only sees stale values while iterating, but the remove()will delete the live entry.

问题是在我迭代时另一个线程可能正在更新或修改值。如果发生这种情况，这些更新可能会永远丢失，因为我的线程在迭代时只会看到陈旧的值，但remove()会删除实时条目。

After some consideration, I came up with this workaround:

经过一番考虑，我想出了这个解决方法：

map.forEach((key, value) -> {
    // delete if value is up to date, otherwise leave for next round
    if (map.remove(key, value)) {
        // do something
    }
});

One problem with this is that it won't catch modifications to mutable values that don't implement equals()(such as AtomicInteger). Is there a better way to safely remove with concurrent modifications?

这样做的一个问题是它不会捕获对未实现的可变值equals()（例如AtomicInteger）的修改。有没有更好的方法来安全删除并发修改？

采纳答案by xz2145

Your workaround works but there is one potential scenario. If certain entries have constant updates map.remove(key,value) may never return true until updates are over.

您的解决方法有效，但存在一种潜在情况。如果某些条目有持续更新 map.remove(key,value) 可能永远不会返回 true 直到更新结束。

If you use JDK8 here is my solution

如果您使用 JDK8，这是我的解决方案

for (Iterator<Entry<Integer, Integer>> iter = map.entrySet().iterator(); iter.hasNext(); ) {
    Entry<Integer, Integer> entry = iter.next();
    Map.compute(entry.getKey(), (k, v) -> f(v));
    //do something for prevValue
}
....
private Integer prevValue;

private Integer f(Integer v){
    prevValue = v;
    return null;
}

compute() will apply f(v) to the value and in our case assign the value to the global variable and remove the entry.

compute() 将 f(v) 应用于该值，在我们的例子中，将值分配给全局变量并删除条目。

According to Javadoc it is atomic.

根据 Javadoc，它是原子的。

Attempts to compute a mapping for the specified key and its current mapped value (or null if there is no current mapping). The entire method invocation is performed atomically. Some attempted update operations on this map by other threads may be blocked while computation is in progress, so the computation should be short and simple, and must not attempt to update any other mappings of this Map.

尝试计算指定键及其当前映射值的映射（如果没有当前映射，则为 null）。整个方法调用以原子方式执行。其他线程在这个映射上尝试的一些更新操作在计算过程中可能会被阻塞，所以计算应该是简短的，并且不能尝试更新这个 Map 的任何其他映射。

回答by Dimitar Dimitrov

Your workaround is actually pretty good. There are other facilities on top of which you can build a somewhat similar solution (e.g. using computeIfPresent()and tombstone values), but they have their own caveats and I have used them in slightly different use-cases.

您的解决方法实际上非常好。还有其他设施，您可以在其上构建一个有点相似的解决方案（例如 usingcomputeIfPresent()和 tombstone 值），但它们有自己的警告，我在略有不同的用例中使用了它们。

As for using a type that doesn't implement equals()for the map values, you can use your own wrapper on top of the corresponding type. That's the most straightforward way to inject custom semantics for object equality into the atomic replace/remove operations provided by ConcurrentMap.

至于使用未实现equals()映射值的类型，您可以在相应类型的顶部使用自己的包装器。这是将对象相等的自定义语义注入ConcurrentMap.

Update

更新

Here's a sketch that shows how you can build on top of the ConcurrentMap.remove(Object key, Object value)API:

这是一个草图，展示了如何在ConcurrentMap.remove(Object key, Object value)API之上构建：

• Define a wrapper type on top of the mutable type you use for the values, also defining your custom equals()method building on top of the current mutable value.
• In your BiConsumer(the lambda you're passing to forEach), create a deep copy of the value (which is of type your new wrapper type) and perform your logic determining whether the value needs to be removed on the copy.
• If the value needs to be removed, call remove(myKey, myValueCopy).
  • If there have been some concurrent changes while you were calculating whether the value needs to be removed, remove(myKey, myValueCopy)will return false(barring ABA problems, which are a separate topic).

• 在用于值的可变类型之上定义一个包装器类型，同时定义equals()在当前可变值之上构建的自定义方法。
• 在您的BiConsumer（您传递给的 lambda 表达式forEach）中，创建该值的深层副本（属于您的新包装器类型）并执行您的逻辑，以确定是否需要在副本上删除该值。
• 如果需要删除该值，请调用remove(myKey, myValueCopy)。
  • 如果在计算是否需要删除值时发生了一些并发更改，remove(myKey, myValueCopy)将返回false（除了 ABA 问题，这是一个单独的主题）。

Here's some code illustrating this:

这是一些说明这一点的代码：

import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

public class Playground {

   private static class AtomicIntegerWrapper {
      private final AtomicInteger value;

      AtomicIntegerWrapper(int value) {
         this.value = new AtomicInteger(value);
      }

      public void set(int value) {
         this.value.set(value);
      }

      public int get() {
         return this.value.get();
      }

      @Override
      public boolean equals(Object obj) {
         if (this == obj) {
            return true;
         }
         if (!(obj instanceof AtomicIntegerWrapper)) {
            return false;
         }
         AtomicIntegerWrapper other = (AtomicIntegerWrapper) obj;
         if (other.value.get() == this.value.get()) {
            return true;
         }
         return false;
      }

      public static AtomicIntegerWrapper deepCopy(AtomicIntegerWrapper wrapper) {
         int wrapped = wrapper.get();
         return new AtomicIntegerWrapper(wrapped);
      }
   }

   private static final ConcurrentMap<Integer, AtomicIntegerWrapper> MAP
         = new ConcurrentHashMap<>();

   private static final int NUM_THREADS = 3;

   public static void main(String[] args) throws InterruptedException {
      for (int i = 0; i < 10; ++i) {
         MAP.put(i, new AtomicIntegerWrapper(1));
      }

      Thread.sleep(1);

      for (int i = 0; i < NUM_THREADS; ++i) {
         new Thread(() -> {
            Random rnd = new Random();
            while (!MAP.isEmpty()) {
               MAP.forEach((key, value) -> {
                  AtomicIntegerWrapper elem = MAP.get(key);
                  if (elem == null) {
                     System.out.println("Oops...");
                  } else if (elem.get() == 1986) {
                     elem.set(1);
                  } else if ((rnd.nextInt() & 128) == 0) {
                     elem.set(1986);
                  }
               });
            }
         }).start();
      }

      Thread.sleep(1);

      new Thread(() -> {
         Random rnd = new Random();
         while (!MAP.isEmpty()) {
            MAP.forEach((key, value) -> {
               AtomicIntegerWrapper elem =
                     AtomicIntegerWrapper.deepCopy(MAP.get(key));
               if (elem.get() == 1986) {
                  try {
                     Thread.sleep(10);
                  } catch (Exception e) {}
                  boolean replaced = MAP.remove(key, elem);
                  if (!replaced) {
                     System.out.println("Bailed out!");
                  } else {
                     System.out.println("Replaced!");
                  }
               }
            });
         }
      }).start();
   }
}

You'll see printouts of "Bailed out!", intermixed with "Replaced!" (removal was successful, as there were no concurrent updates that you care about) and the calculation will stop at some point.

您会看到“Bailed out！”的打印输出，与“Replaced！”混杂在一起。（删除成功，因为没有您关心的并发更新）并且计算将在某个时候停止。

• If you remove the custom equals()method and continue to use a copy, you'll see an endless stream of "Bailed out!", because the copy is never considered equal to the value in the map.
• If you don't use a copy, you won't see "Bailed out!" printed out, and you'll hit the problem you're explaining - values are removed regardless of concurrent changes.

• 如果删除自定义equals()方法并继续使用副本，您将看到源源不断的“Bailed out！”，因为副本永远不会被视为等于映射中的值。
• 如果您不使用副本，您将不会看到“Bailed out！” 打印出来，你会遇到你正在解释的问题 - 无论并发更改如何，值都会被删除。

回答by dieter

Let us consider what options you have.

让我们考虑一下您有哪些选择。

1. Create your own Container-class with isUpdated()operation and use your own workaround.

2. If your map contains just a few elements and you are iterating over the map very frequently compared against put/delete operation. It could be a good choice to use CopyOnWriteArrayListCopyOnWriteArrayList<Entry<Integer, Integer>> lookupArray = ...;

3. The other option is to implement your own CopyOnWriteMap

   public class CopyOnWriteMap<K, V> implements Map<K, V>{
   
       private volatile Map<K, V> currentMap;
   
       public V put(K key, V value) {
           synchronized (this) {
               Map<K, V> newOne = new HashMap<K, V>(this.currentMap);
               V val = newOne.put(key, value);
               this.currentMap = newOne; // atomic operation
               return val;
           }
       }
   
       public V remove(Object key) {
           synchronized (this) {
               Map<K, V> newOne = new HashMap<K, V>(this.currentMap);
               V val = newOne.remove(key);
               this.currentMap = newOne; // atomic operation
               return val;
           }
       }
   
       [...]
   }
   

1. 使用isUpdated()操作创建您自己的容器类并使用您自己的解决方法。

2. 如果您的地图仅包含几个元素，并且与放置/删除操作相比，您非常频繁地迭代地图。使用它可能是一个不错的选择CopyOnWriteArrayListCopyOnWriteArrayList<Entry<Integer, Integer>> lookupArray = ...;

3. 另一种选择是实现你自己的 CopyOnWriteMap

   public class CopyOnWriteMap<K, V> implements Map<K, V>{
   
       private volatile Map<K, V> currentMap;
   
       public V put(K key, V value) {
           synchronized (this) {
               Map<K, V> newOne = new HashMap<K, V>(this.currentMap);
               V val = newOne.put(key, value);
               this.currentMap = newOne; // atomic operation
               return val;
           }
       }
   
       public V remove(Object key) {
           synchronized (this) {
               Map<K, V> newOne = new HashMap<K, V>(this.currentMap);
               V val = newOne.remove(key);
               this.currentMap = newOne; // atomic operation
               return val;
           }
       }
   
       [...]
   }
   

There is a negative side effect. If you are using copy-on-write Collections your updates will be never lost, but you can see some former deleted entry again.

有一个负面的副作用。如果您使用写时复制集合，您的更新将永远不会丢失，但您可以再次看到一些以前删除的条目。

Worst case: deleted entry will be restored every time if map get copied.

最坏的情况：如果地图被复制，每次删除的条目都会恢复。