Iterate über ConcurrentHashMap während Einträge

Question

ich in regelmäßigen Abständen wollen Löschen einen ConcurrentHashMap iterieren während Einträge zu entfernen, wie folgt aus:

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

Das Problem ist, dass ein anderer Thread aktualisieren können oder Werte zu modifizieren, während ich laufen. Wenn das passiert, können diese Updates für immer verloren gehen, weil mein Thread nur veraltete Werte während der Iteration sieht, aber der remove() wird den Live-Eintrag löschen.

Nach einiger Überlegung kam ich mit dieser Problemumgehung oben:

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

Ein Problem dabei ist, dass es keine Änderungen an veränderbaren Werte fangen, die equals() (wie AtomicInteger) nicht implementieren. Gibt es einen besseren Weg, um mit gleichzeitigen Änderungen sicher zu entfernen?

Answer 1

Ihre Problemumgehung funktioniert, aber es gibt ein mögliches Szenario. Wenn bestimmte Einträge konstante Aktualisierungen aufweisen, wird map.remove (Schlüssel, Wert) möglicherweise nie den Wert true zurückgeben, bis die Aktualisierungen abgeschlossen sind.

Wenn Sie JDK8 hier verwenden ist meine Lösung

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() wird f (v) auf den Wert und in unserem Fall weisen Sie den Wert auf die globale Variable übernehmen und den Eintrag löschen.

Laut Javadoc ist es atomar.

Es wird versucht, eine Zuordnung für den angegebenen Schlüssel und den aktuellen zugeordneten Wert zu berechnen (oder null, wenn keine aktuelle Zuordnung vorhanden ist). Der gesamte Methodenaufruf wird atomar ausgeführt. Einige versuchte Aktualisierungsoperationen in dieser Map durch andere Threads werden möglicherweise während der Berechnung blockiert. Daher sollte die Berechnung kurz und einfach sein und nicht versuchen, andere Zuordnungen dieser Map zu aktualisieren.

Answer 2

Ihr Workaround ist eigentlich ziemlich gut. Es gibt andere Möglichkeiten, auf denen Sie eine ähnliche Lösung (z. B. computeIfPresent() und Tombstone-Werte) bauen können, aber sie haben ihre eigenen Vorbehalte und ich habe sie in etwas anderen Anwendungsfällen verwendet.

Wenn Sie einen Typ verwenden, der equals() für die Kartenwerte nicht implementiert, können Sie einen eigenen Wrapper über dem entsprechenden Typ verwenden. Das ist die einfachste Möglichkeit, benutzerdefinierte Semantiken für die Objektgleichheit in die von ConcurrentMap bereitgestellten atomaren Ersetzungs-/Entfernungsoperationen zu injizieren.

aktualisieren

Hier ist eine Skizze, die zeigt, wie Sie auf dem ConcurrentMap.remove(Object key, Object value) API bauen können:

• definieren einen Wrapper-Typ auf dem wandelbaren Typ Sie für die Werte verwenden, auch Definition Ihrer benutzerdefinierten equals() Methode, die auf dem aktuellen veränderbaren Wert basiert.
• Erstellen Sie in Ihrem BiConsumer (das Lambda, das Sie an forEach übergeben) eine tiefe Kopie des Werts (der Ihren neuen Wrappertyp aufweist) und führen Sie Ihre Logik durch, um zu bestimmen, ob der Wert auf der Kopie entfernt werden muss. Wenn der Wert entfernt werden muss, wählen Sie remove(myKey, myValueCopy).
  • Wenn es einige gleichzeitige Änderungen vorgenommen wurden, während Sie wurden der Berechnung, ob der Wert entfernt werden muss, wird remove(myKey, myValueCopy) zurückkehren false (abgesehen von ABA Probleme, die ein separates Thema sind).

Hier einige Code dies veranschaulicht: "Ersetzt"

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(); 
    } 
} 

Sie Ausdrucke von "! Gerettet" sehen werden, vermischt mit (Die Entfernung war erfolgreich, da es keine gleichzeitigen Aktualisierungen gab, die Sie interessieren) und die Berechnung wird irgendwann aufhören.

• Wenn Sie die benutzerdefinierten equals() Methode entfernen und weiterhin eine Kopie verwenden, werden Sie einen endlosen Strom von „Klemme geholfen!“, Weil die Kopie nie auf den Wert in der Karte gleich betrachtet wird.
• Wenn Sie keine Kopie verwenden, sehen Sie nicht "Bailed out!" ausgedruckt, und Sie werden das Problem, das Sie erklären, treffen - Werte werden unabhängig von gleichzeitigen Änderungen entfernt.

Answer 3

Lassen Sie uns überlegen, welche Optionen Sie haben.

1. Erstellen Sie Ihre eigene Container-Klasse mit isUpdated() Betrieb und Ihre eigene Abhilfe verwenden.

2. Wenn Ihre Map nur ein paar Elemente enthält und Sie sehr häufig über die Map iterieren, vergleichen Sie sie mit put/delete. Es könnte eine gute Wahl CopyOnWriteArrayList CopyOnWriteArrayList<Entry<Integer, Integer>> lookupArray = ...;

3. Die andere Option ist zu implementieren Ihre eigenen 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; 
        } 
       } 
   
       [...] 
   } 
   

Es ist ein negativer Nebeneffekt zu verwenden sein. Wenn Sie Copy-on-Write-Sammlungen verwenden, gehen Ihre Aktualisierungen nie verloren, aber Sie können wieder einige frühere gelöschte Einträge sehen.

Worst case: Der gelöschte Eintrag wird jedes Mal wiederhergestellt, wenn die Karte kopiert wird.