ConcurrentLinkedDeque in Java (mit Beispiel)

In diesem Artikel erfährst du alles über die Klasse java.util.concurrent.ConcurrentLinkedDeque:

• Was sind die Eigenschaften des ConcurrentLinkedDeque?
• Wann sollte man es einsetzen?
• Wie setzt man es ein (Java-Beispiel)?

Hier befinden wir uns in der Klassenhierarchie:

ConcurrentLinkedDeque Eigenschaften

ConcurrentLinkedDeque ist das Deque-Pendant zur ConcurrentLinkedQueue und teilt deren Eigenschaften:

• Es basiert auf einer doppelt verketteten Liste.
• Threadsicherheit wird garantiert durch optimistisches Locking in Form von nicht-blockierenden Compare-and-set (CAS)-Operationen auf separate VarHandles für Kopf und Ende des Deques, sowie für die Referenzen der Listenknoten.
• Um die Länge eines ConcurrentLinkedDeque zu bestimmen, müssen die Elemente der verketteten Liste gezählt werden. Der Aufwand hierfür wächst proportional mit der Länge der Liste. Die Zeitkomplexität dafür beträgt also: O(n)
• Aufgrund der hohen Kosten für die Längenbestimmung ist ConcurrentLinkedDeque unbounded.

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: Alle Elemente, die zum Zeitpunkt der Erzeugung des Interators im Deque liegen, werden vom Iterator genau einmal durchlaufen. Änderungen, die danach erfolgen, können – müssen aber nicht – durch den Iterator berücksichtigt werden.

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 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 Seiten 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;

  public static void main(String[] args) 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 Seiten 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.

Zusammenfassung und Ausblick

In diesem Teil der Tutorialserie hast du das auf einer verketteten Liste basierte, threadsichere ConcurrentLinkedDeque und seine Eigenschaften kennengelernt.

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