ConcurrentLinkedDeque in Java: threadsicher ohne Lock

In diesem Artikel erfährst du alles über die Klasse java.util.concurrent.ConcurrentLinkedDeque, die es seit Java 7 gibt:

• Was sind die Eigenschaften des ConcurrentLinkedDeque?
• Wann sollte man es einsetzen – und wann die ConcurrentLinkedQueue?
• Wie setzt man es ein (Java-Beispiel)?
• Wie erreicht es Threadsicherheit ohne Lock?

Hier befinden wir uns in der Klassenhierarchie:

Seit Java 21 erweitert Deque außerdem das Interface SequencedCollection. Damit hat auch ConcurrentLinkedDeque die Methode reversed(), die eine umgekehrte Sicht auf das Deque liefert. Was das für ein Deque bedeutet, erfährst du im Artikel Queue vs. Deque.

ConcurrentLinkedDeque Eigenschaften

ConcurrentLinkedDeque ist das Deque-Pendant zur ConcurrentLinkedQueue und teilt deren Eigenschaften bis auf eine: Es basiert auf einer doppelt verketteten Liste. Jeder Knoten hat neben der Referenz auf seinen Nachfolger (next) auch eine auf seinen Vorgänger (prev). Damit lassen sich an jedem der beiden Enden Elemente sowohl einfügen als auch entnehmen.

ConcurrentLinkedDeque ist unbounded: Eine Kapazitätsgrenze bräuchte einen Zähler, der bei jedem Einfügen und Entnehmen zusammen mit der Liste aktualisiert wird. LinkedBlockingDeque führt genau so einen Zähler; der lock-freie Algorithmus des ConcurrentLinkedDeque spart ihn sich. Blockierende Operationen bietet ConcurrentLinkedDeque ebenfalls nicht.

null-Elemente sind nicht erlaubt (offerFirst(null) wirft eine NullPointerException) – deshalb können pollFirst() und pollLast() mit dem Rückgabewert null eindeutig „Deque leer“ signalisieren.

Die Eigenschaften im Detail:

Unterliegende Datenstruktur	Thread-safe?	Blocking/ Non-blocking	Bounded/ Unbounded	Iterator Type
Doppelt verkettete Liste	Ja (optimistisches Locking durch Compare-and-set)	Non-blocking	Unbounded	Weakly consistent¹

¹ Weakly consistent: Elemente, die seit der Erzeugung des Iterators ununterbrochen im Deque liegen, liefert der Iterator genau einmal. Änderungen nach der Erzeugung – Einfügungen wie Entnahmen – kann er berücksichtigen, muss es aber nicht; eine ConcurrentModificationException wirft er nie.

Der Iterator unterstützt remove().

Die einzelnen Operationen an den beiden Enden – offerFirst(), offerLast(), pollFirst() und pollLast() – sind atomar: Jede von ihnen wird für andere Threads in genau einem Moment wirksam, dem erfolgreichen Compare-and-set. Bulk-Operationen wie addAll(), removeIf() und forEach() sind es nicht: Ein forEach() parallel zu addAll() sieht unter Umständen nur einen Teil der neuen Elemente.

Einsatzempfehlung

ConcurrentLinkedDeque ist eine gute Wahl, wenn ein threadsicheres, nicht blockierendes, unbounded Deque benötigt wird.

Für diesen Einsatzzweck existiert keine Array-basierte Alternative. Das einzige Array-basierte Deque, ArrayDeque, ist nicht threadsicher.

ConcurrentLinkedDeque vs. ConcurrentLinkedQueue

Brauchst du nur eine FIFO-Queue – einfügen an einem Ende, entnehmen am anderen –, empfehle ich die ConcurrentLinkedQueue. Sie arbeitet nach demselben lock-freien Prinzip, kommt aber mit einer Referenz pro Knoten weniger aus, denn ihre Liste ist nur einfach verkettet.

ConcurrentLinkedDeque brauchst du, sobald du an beiden Enden einfügen oder entnehmen musst – etwa für einen threadsicheren Stack mit push() und pop() oder für eine Warteschlange, in der dringende Aufträge mit offerFirst() vorne eingereiht werden.

ConcurrentLinkedDeque Beispiel

Das folgende Beispiel (Klasse ConcurrentLinkedDequeExample auf GitHub) demonstriert die Threadsicherheit von ConcurrentLinkedDeque. Vier schreibende und drei lesende Threads fügen nebenläufig Elemente an zufälligen Enden des Deques ein bzw. entnehmen sie.

public class ConcurrentLinkedDequeExample {

  private static final int NUMBER_OF_PRODUCERS = 4;
  private static final int NUMBER_OF_CONSUMERS = 3;
  private static final int NUMBER_OF_ELEMENTS_TO_PUT_INTO_DEQUE_PER_THREAD = 5;
  private static final int MIN_SLEEP_TIME_MILLIS = 500;
  private static final int MAX_SLEEP_TIME_MILLIS = 2000;

  private Deque<Integer> deque;
  private final CountDownLatch producerFinishLatch =
      new CountDownLatch(NUMBER_OF_PRODUCERS);
  private volatile boolean consumerShouldBeStoppedWhenDequeIsEmpty;

  static void main() throws InterruptedException {
    new ConcurrentLinkedDequeExample().runDemo();

    // We'll let the program end when all consumers are finished
  }

  private void runDemo() throws InterruptedException {
    createDeque();
    startProducers();
    startConsumers();
    waitUntilAllProducersAreFinished();

    consumerShouldBeStoppedWhenDequeIsEmpty = true;
  }

  private void createDeque() {
    deque = new ConcurrentLinkedDeque<>();
  }

  private void startProducers() {
    for (int i = 0; i < NUMBER_OF_PRODUCERS; i++) {
      createProducerThread().start();
    }
  }

  private Thread createProducerThread() {
    return new Thread(
        () -> {
          for (int i = 0; i < NUMBER_OF_ELEMENTS_TO_PUT_INTO_DEQUE_PER_THREAD; i++) {
            sleepRandomTime();
            insertRandomElementAtRandomSide();
          }

          producerFinishLatch.countDown();
        });
  }

  private void sleepRandomTime() {
    ThreadLocalRandom random = ThreadLocalRandom.current();
    try {
      Thread.sleep(random.nextInt(MIN_SLEEP_TIME_MILLIS, MAX_SLEEP_TIME_MILLIS));
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  private void insertRandomElementAtRandomSide() {
    ThreadLocalRandom random = ThreadLocalRandom.current();
    Integer element = random.nextInt(1000);
    if (random.nextBoolean()) {
      deque.offerFirst(element);
      System.out.printf(
          "[%s] deque.offerFirst(%3d)        --> deque = %s%n",
          Thread.currentThread().getName(), element, deque);
    } else {
      deque.offerLast(element);
      System.out.printf(
          "[%s] deque.offerLast(%3d)         --> deque = %s%n",
          Thread.currentThread().getName(), element, deque);
    }
  }

  private void startConsumers() {
    for (int i = 0; i < NUMBER_OF_CONSUMERS; i++) {
      createConsumerThread().start();
    }
  }

  private Thread createConsumerThread() {
    return new Thread(
        () -> {
          while (shouldConsumerContinue()) {
            sleepRandomTime();
            removeElementFromRandomSide();
          }
        });
  }

  private boolean shouldConsumerContinue() {
    return !(consumerShouldBeStoppedWhenDequeIsEmpty && deque.isEmpty());
  }

  private void removeElementFromRandomSide() {
    if (ThreadLocalRandom.current().nextBoolean()) {
      Integer element = deque.pollFirst();
      System.out.printf(
          "[%s]     deque.pollFirst() = %4d --> deque = %s%n",
          Thread.currentThread().getName(), element, deque);
    } else {
      Integer element = deque.pollLast();
      System.out.printf(
          "[%s]     deque.pollLast()  = %4d --> deque = %s%n",
          Thread.currentThread().getName(), element, deque);
    }
  }

  private void waitUntilAllProducersAreFinished() throws InterruptedException {
    producerFinishLatch.await();
  }
}

Im Folgenden habe ich die ersten 15 Zeilen eines beispielhaften Programmlaufs abgedruckt:

[Thread-1] deque.offerFirst(295)        --> deque = [295]
[Thread-4]     deque.pollLast()  =  295 --> deque = []
[Thread-5]     deque.pollLast()  = null --> deque = []
[Thread-2] deque.offerLast(982)         --> deque = [982]
[Thread-3] deque.offerFirst(190)        --> deque = [190, 982]
[Thread-0] deque.offerFirst(522)        --> deque = [522, 190, 982]
[Thread-6]     deque.pollLast()  =  982 --> deque = [522, 190]
[Thread-1] deque.offerLast(543)         --> deque = [522, 190, 543]
[Thread-0] deque.offerFirst(506)        --> deque = [506, 522, 190, 543]
[Thread-5]     deque.pollLast()  =  543 --> deque = [506, 522, 190]
[Thread-4]     deque.pollFirst() =  506 --> deque = [522, 190]
[Thread-3] deque.offerLast(760)         --> deque = [522, 190, 760]
[Thread-2] deque.offerFirst( 46)        --> deque = [46, 522, 190, 760]
[Thread-6]     deque.pollLast()  =  760 --> deque = [46, 522, 190]
[Thread-1] deque.offerLast(312)         --> deque = [46, 522, 190, 312]

Es ist schön zu sehen, wie die sieben Threads Elemente an beiden Enden des Deques einfügen und entnehmen. In der dritten Zeile siehst du, wie Thread 5 beim Aufruf von pollLast() den Rückgabewert null geliefert bekommen hat, da zu diesem Zeitpunkt das Deque leer war.

ConcurrentLinkedDeque Performance

Dieser Abschnitt behandelt die Threadsicherheit und Zeitkomplexität von ConcurrentLinkedDeque.

Ist ConcurrentLinkedDeque threadsicher?

Ja. Die Threadsicherheit des ConcurrentLinkedDeque wird – wie bei der ConcurrentLinkedQueue – durch optimistisches Locking erzielt, und auch hier führt das Wort „Locking“ in die Irre: Die Klasse kommt ganz ohne Lock aus. Sie arbeitet mit nicht-blockierenden Compare-and-set-Operationen (CAS):

• offerFirst() hängt den neuen Knoten per Compare-and-set an die prev-Referenz des ersten Knotens, offerLast() an die next-Referenz des letzten Knotens.
• pollFirst() und pollLast() setzen per Compare-and-set das item des ersten bzw. letzten Knotens auf null. Damit ist das Element entnommen; aus der Liste ausgehängt wird der Knoten erst danach.

Die Zeiger head und tail dürfen dabei hinterherhinken. Ohne Lock lässt sich das gar nicht vermeiden: Ein Compare-and-set ändert nur eine Referenz, beim Einfügen ändern sich aber zwei – die am Randknoten und der Zeiger head bzw. tail. Zwischen den beiden Schritten zeigt der Zeiger für andere Threads noch auf den bisherigen Randknoten.

Die Klasse macht daraus einen Vorteil: Ein Thread zieht head bzw. tail nur nach, wenn er auf dem Weg zum ersten bzw. letzten Knoten über mindestens einen Knoten hinweggehen musste. Ohne Konkurrenz geschieht das nur bei jedem zweiten Einfügen bzw. Entnehmen – die Hälfte der Compare-and-sets auf den beiden Zeigern entfällt, und um genau diese Felder konkurrieren alle Threads, die am selben Ende arbeiten.

Ein Thread verlässt sich deshalb nicht auf die beiden Zeiger, sondern prüft am Knoten selbst – prev == null heißt erster Knoten, next == null heißt letzter Knoten – und folgt sonst den prev- bzw. next-Referenzen weiter.

Was passiert, wenn zwei Threads gleichzeitig an verschiedenen Enden einfügen, zeigt das folgende Diagramm mit Werten aus den Zeilen 7 bis 9 der Beispielausgabe: Thread-0 setzt sein Compare-and-set auf prev des ersten Knotens 522 ab, Thread-1 seines auf next des letzten Knotens 190. Die beiden Compare-and-sets treffen verschiedene Referenzen, deshalb gelingen beide, und kein Thread muss seinen Versuch wiederholen. head und tail bleiben dabei, wo sie waren, denn keiner der beiden Threads musste über einen Knoten hinweggehen.

Um dieselbe Referenz konkurrieren nur Threads, die am selben Ende einfügen. Dann gelingt – wie bei der ConcurrentLinkedQueue – nur ein Compare-and-set, und der andere Thread wiederholt seinen Versuch am neuen Ende des Deques.

Beim Zugriff auf Deques ist normalerweise niedrige bis moderate Contention (Zugriffskonflikte durch mehrere Threads) zu erwarten, da ein Thread in der Regel nicht ununterbrochen auf das Deque zugreift, sondern das einzufügende Element erst erzeugen bzw. das entnommene Element verarbeiten muss. Bei niedriger bis moderater Contention erzielt optimistisches Locking einen deutlichen Performance-Gewinn gegenüber pessimistischem Locking durch implizite oder explizite Locks.

Bei LinkedBlockingDeque hingegen wird die Threadsicherheit durch ein einzelnes ReentrantLock gewährleistet, das beide Enden gemeinsam schützt.

ConcurrentLinkedDeque Zeitkomplexität

Wie bei allen Deques auf Basis verketteter Listen ist der Aufwand für das Einfügen und Entnehmen an beiden Enden unabhängig von der Länge des Deques. Die Zeitkomplexität beträgt also O(1).

Dies gilt hingegen nicht für die size()-Methode. Um die Länge des Deques zu bestimmen, muss über alle Elemente der verketteten Liste iteriert werden. Je länger das Deque, desto länger dauert es, die Länge zu berechnen. Die Zeitkomplexität für size() beträgt somit O(n).

Und weil andere Threads währenddessen einfügen und entnehmen dürfen, ist das Ergebnis im Moment der Rückgabe unter Umständen schon überholt – das Javadoc hält size() in nebenläufigen Anwendungen deshalb für wenig nützlich. Für die Frage „Ist das Deque leer?“ nimmst du isEmpty(): Es sucht nur das erste Element und zählt nicht weiter.

(Hier findest du eine Einführung in das Thema Zeitkomplexität.)

Zusammenfassung und Ausblick

In diesem Teil der Tutorialserie hast du das auf einer doppelt verketteten Liste basierende, threadsichere ConcurrentLinkedDeque kennengelernt: lock-frei per Compare-and-set, unbounded, mit size() in O(n). Brauchst du nur eine FIFO-Queue, nimmst du die ConcurrentLinkedQueue.

Im nächsten Teil stelle ich dir das einzige blockierende Deque, LinkedBlockingDeque, vor.