{"id":29066,"date":"2022-04-20T17:06:28","date_gmt":"2022-04-20T15:06:28","guid":{"rendered":"https:\/\/www.happycoders.eu\/?p=29066"},"modified":"2024-11-27T15:09:28","modified_gmt":"2024-11-27T14:09:28","slug":"linkedblockingqueue-java","status":"publish","type":"post","link":"https:\/\/www.happycoders.eu\/de\/algorithmen\/linkedblockingqueue-java\/","title":{"rendered":"LinkedBlockingQueue in Java (mit Beispiel)"},"content":{"rendered":"\n

In diesem Teil der Tutorialserie geht es um die LinkedBlockingQueue<\/code>. Du wirst deren speziellen Eigenschaften kennenlernen und anhand eines Beispiels sehen, wie man diese Queue einsetzt. Au\u00dferdem wirst du erfahren, wann du genau diese Queue einsetzen solltest.<\/p>\n\n\n\n

Hier befinden wir uns in der Klassenhierarchie:<\/p>\n\n\n\n

\"LinkedBlockingQueue
LinkedBlockingQueue in der Klassenhierarchie<\/figcaption><\/figure><\/div>\n\n\n\n

LinkedBlockingQueue Eigenschaften<\/h2>\n\n\n\n

Die Klasse java.util.concurrent<\/code>.LinkedBlockingQueue<\/code> basiert \u2013 genau wie ConcurrentLinkedQueue<\/a> \u2013 auf einer verketteten Liste, ist allerdings \u2013 ebenso wie die im n\u00e4chsten Teil vorgestellte ArrayBlockingQueue<\/a> \u2013 threadsicher (s. u.<\/a>), bounded und blockierend.<\/p>\n\n\n\n

Anders als die ArrayBlockingQueue<\/code> bietet LinkedBlockingQueue<\/code> keine Fairness Policy an. (Fairness Policy bedeutet, dass blockierende Methoden in der Reihenfolge bedient werden, in der sie aufgerufen werden.)<\/p>\n\n\n\n

Die Queue-Eigenschaften im Detail:<\/p>\n\n\n\n

Unterliegende Datenstruktur<\/th>Thread-safe?<\/th>Blocking\/
Non-blocking<\/th>		Fairness
Policy<\/th>		Bounded\/
Unbounded<\/th>		Iterator Type<\/th><\/tr><\/thead>
Verkettete Liste<\/td>Ja
(pessimistisches Locking mit zwei Locks)<\/span><\/td>		Blocking<\/td>Nicht verf\u00fcgbar<\/td>Bounded<\/td>Weakly consistent\u00b9<\/td><\/tr><\/tbody><\/table><\/figure>\n\n\n\n

\u00b9 Weakly consistent: Alle Elemente, die zum Zeitpunkt der Erzeugung des Interators in der Queue liegen, werden vom Iterator genau einmal durchlaufen. \u00c4nderungen, die danach erfolgen, k\u00f6nnen \u2013 m\u00fcssen aber nicht \u2013 durch den Iterator ber\u00fccksichtigt werden.<\/p>\n\n\n\n

Einsatzempfehlung<\/h3>\n\n\n\n

Ich empfehle LinkedBlockingQueue<\/code>, wenn du eine blockierende, threadsichere Queue ben\u00f6tigst.<\/p>\n\n\n\n

Die Klasse LinkedBlockingQueue<\/code> wird \u00fcbrigens von Executors.newFixedThreadPool()<\/code> und Executors.newSingleThreadedExecutor()<\/code> als \"work queue\" f\u00fcr den Executor verwendet. Sie wird somit intensiv genutzt, was die Wahrscheinlichkeit f\u00fcr Bugs \u00e4u\u00dferst gering h\u00e4lt.<\/p>\n\n\n\n

LinkedBlockingQueue Beispiel<\/h2>\n\n\n\n

Das folgende Beispiel zeigt, wie die LinkedBlockingQueue<\/code> verwendet wird. Wir erzeugen eine Queue mit der Kapazit\u00e4t 3. Direkt im Anschluss beginnen wir im Abstand von jeweils drei Sekunden Elemente aus der Queue zu lesen. Erst nach 3,5 Sekunden beginnen wir im Abstand von jeweils einer Sekunde Elemente in die Queue zu schreiben (\u2192 Code auf GitHub<\/a>).<\/p>\n\n\n

public<\/span> class<\/span> LinkedBlockingQueueExample<\/span> <\/span>{\n  private<\/span> static<\/span> final<\/span> long<\/span> startTime = System.currentTimeMillis();\n\n  public<\/span> static<\/span> void<\/span> main<\/span>(String[] args)<\/span> throws<\/span> InterruptedException <\/span>{\n    BlockingQueue<Integer> queue = new<\/span> LinkedBlockingQueue<>(3<\/span>);\n    ScheduledExecutorService pool = Executors.newScheduledThreadPool(10<\/span>);\n\n    \/\/ Start reading from the queue immediately, every 3 seconds<\/span>\n    for<\/span> (int<\/span> i = 0<\/span>; i < 10<\/span>; i++) {\n      int<\/span> delaySeconds = i * 3<\/span>;\n      pool.schedule(() -> dequeue(queue), delaySeconds, TimeUnit.SECONDS);\n    }\n\n    \/\/ Start writing to the queue after 3.5 seconds (so there are already 2 threads <\/span>\n    \/\/ waiting), every 1 seconds (so that the queue fills faster than it's emptied, <\/span>\n    \/\/ so that we see a full queue soon)<\/span>\n    for<\/span> (int<\/span> i = 0<\/span>; i < 10<\/span>; i++) {\n      int<\/span> element = i; \/\/ Assign to an effectively final variable<\/span>\n      int<\/span> delayMillis = 3500<\/span> + i * 1000<\/span>;\n      pool.schedule(() -> enqueue(queue, element), delayMillis, TimeUnit.MILLISECONDS);\n    }\n\n    pool.shutdown();\n    pool.awaitTermination(1<\/span>, TimeUnit.MINUTES);\n  }\n\n  private<\/span> static<\/span> void<\/span> enqueue<\/span>(BlockingQueue<Integer> queue, int<\/span> element)<\/span> <\/span>{\n    log(\"Calling queue.put(%d) (queue = %s)...\"<\/span>, element, queue);\n    try<\/span> {\n      queue.put(element);\n      log(\"queue.put(%d) returned (queue = %s)\"<\/span>, element, queue);\n    } catch<\/span> (InterruptedException e) {\n      Thread.currentThread().interrupt();\n    }\n  }\n\n  private<\/span> static<\/span> void<\/span> dequeue<\/span>(BlockingQueue<Integer> queue)<\/span> <\/span>{\n    log(\"    Calling queue.take() (queue = %s)...\"<\/span>, queue);\n    try<\/span> {\n      Integer element = queue.take();\n      log(\"    queue.take() returned %d (queue = %s)\"<\/span>, element, queue);\n    } catch<\/span> (InterruptedException e) {\n      Thread.currentThread().interrupt();\n    }\n  }\n\n  private<\/span> static<\/span> void<\/span> log<\/span>(String format, Object... args)<\/span> <\/span>{\n    System.out.printf(\n        Locale.US,\n        \"[%4.1fs] [%-16s] %s%n\"<\/span>,\n        (System.currentTimeMillis() - startTime) \/ 1000.0<\/span>,\n        Thread.currentThread().getName(),\n        String.format(format, args));\n  }\n}<\/code><\/span>Code-Sprache:<\/span> Java<\/span> (<\/span>java<\/span>)<\/span><\/small><\/pre>\n\n\n
Im folgenden siehst du die Ausgabe des Beispiel-Programms:<\/p>\n\n\n
[ 0.0s] [pool-1-thread-1 ]     Calling queue.take() (queue = [])...\n[ 3.0s] [pool-1-thread-4 ]     Calling queue.take() (queue = [])...\n[ 3.5s] [pool-1-thread-8 ] Calling queue.put(0) (queue = [])...\n[ 3.5s] [pool-1-thread-1 ]     queue.take() returned 0 (queue = [])\n[ 3.5s] [pool-1-thread-8 ] queue.put(0) returned (queue = [])\n[ 4.5s] [pool-1-thread-5 ] Calling queue.put(1) (queue = [])...\n[ 4.5s] [pool-1-thread-4 ]     queue.take() returned 1 (queue = [])\n[ 4.5s] [pool-1-thread-5 ] queue.put(1) returned (queue = [])\n[ 5.5s] [pool-1-thread-3 ] Calling queue.put(2) (queue = [])...\n[ 5.5s] [pool-1-thread-3 ] queue.put(2) returned (queue = [2])\n[ 6.0s] [pool-1-thread-7 ]     Calling queue.take() (queue = [2])...\n[ 6.0s] [pool-1-thread-7 ]     queue.take() returned 2 (queue = [])\n[ 6.5s] [pool-1-thread-9 ] Calling queue.put(3) (queue = [])...\n[ 6.5s] [pool-1-thread-9 ] queue.put(3) returned (queue = [3])\n[ 7.5s] [pool-1-thread-6 ] Calling queue.put(4) (queue = [3])...\n[ 7.5s] [pool-1-thread-6 ] queue.put(4) returned (queue = [3, 4])\n[ 8.5s] [pool-1-thread-2 ] Calling queue.put(5) (queue = [3, 4])...\n[ 8.5s] [pool-1-thread-2 ] queue.put(5) returned (queue = [3, 4, 5])\n[ 9.0s] [pool-1-thread-10]     Calling queue.take() (queue = [3, 4, 5])...\n[ 9.0s] [pool-1-thread-10]     queue.take() returned 3 (queue = [4, 5])\n[ 9.5s] [pool-1-thread-1 ] Calling queue.put(6) (queue = [4, 5])...\n[ 9.5s] [pool-1-thread-1 ] queue.put(6) returned (queue = [4, 5, 6])\n[10.5s] [pool-1-thread-8 ] Calling queue.put(7) (queue = [4, 5, 6])...\n[11.5s] [pool-1-thread-4 ] Calling queue.put(8) (queue = [4, 5, 6])...\n[12.0s] [pool-1-thread-5 ]     Calling queue.take() (queue = [4, 5, 6])...\n[12.0s] [pool-1-thread-5 ]     queue.take() returned 4 (queue = [5, 6, 7])\n[12.0s] [pool-1-thread-8 ] queue.put(7) returned (queue = [5, 6, 7])\n[12.5s] [pool-1-thread-3 ] Calling queue.put(9) (queue = [5, 6, 7])...\n[15.0s] [pool-1-thread-7 ]     Calling queue.take() (queue = [5, 6, 7])...\n[15.0s] [pool-1-thread-7 ]     queue.take() returned 5 (queue = [6, 7, 8])\n[15.0s] [pool-1-thread-4 ] queue.put(8) returned (queue = [6, 7, 8])\n[18.0s] [pool-1-thread-9 ]     Calling queue.take() (queue = [6, 7, 8])...\n[18.0s] [pool-1-thread-3 ] queue.put(9) returned (queue = [7, 8, 9])\n[18.0s] [pool-1-thread-9 ]     queue.take() returned 6 (queue = [7, 8, 9])\n[21.0s] [pool-1-thread-6 ]     Calling queue.take() (queue = [7, 8, 9])...\n[21.0s] [pool-1-thread-6 ]     queue.take() returned 7 (queue = [8, 9])\n[24.0s] [pool-1-thread-2 ]     Calling queue.take() (queue = [8, 9])...\n[24.0s] [pool-1-thread-2 ]     queue.take() returned 8 (queue = [9])\n[27.0s] [pool-1-thread-10]     Calling queue.take() (queue = [9])...\n[27.0s] [pool-1-thread-10]     queue.take() returned 9 (queue = [])<\/code><\/span>Code-Sprache:<\/span> Klartext<\/span> (<\/span>plaintext<\/span>)<\/span><\/small><\/pre>\n\n\n
Da wir erst mit dem Schreiben beginnen, nachdem bereits zwei Threads take()<\/code> aufrufen, blockieren diese ersten zwei Leseversuche bei 0,0 und 3,0 s (Thread 1 und 4).<\/p>\n\n\n\n
Nach 3,5 s wird das erste Element geschrieben (Thread 8). Dadurch wird Thread 1 aufgeweckt und die take()<\/code> Methode entnimmt dieses Element sofort wieder aus der Queue. <\/p>\n\n\n\n
Nach 4,5 s wird das zweite Element geschrieben (Thread 5). Thread 4 wird aufgeweckt und entnimmt dieses Element wieder aus der Queue.<\/p>\n\n\n\n
Das Programm schreibt schneller als es liest. Nach 10,5 s blockiert daher zum ersten Mal ein schreibender Thread (Thread 8) beim Versuch die 7 in die - zu dem Zeitpunkt volle - Queue zu schreiben. Nach 11,5 s blockiert ebenfalls Thread 4 beim Versuch die 8 in die Queue zu schreiben.<\/p>\n\n\n\n
Nach 12,0 s entnimmt Thread 5 ein Element aus der Queue. Dadurch wird in der Queue ein Platz frei. Thread 8 wird aufgeweckt und schreibt die 7 in die Queue.    <\/p>\n\n\n\n
Versuch einmal selbst die restlichen Ausgaben zu lesen und zu verstehen.<\/p>\n\n\n\n
Ist LinkedBlockingQueue threadsicher?<\/h2>\n\n\n\n
Ja, LinkedBlockingQueue<\/code> ist threadsicher.<\/p>\n\n\n\n
Die Threadsicherheit der LinkedBlockingQueue<\/code> wird durch pessimistisches Locking mittels zweier separater ReentrantLock<\/code> f\u00fcr Schreib- und Leseoperationen gew\u00e4hrleistet. So kann es zu keiner Contention (Zugriffskonflikten) zwischen Producer- und Consumer-Threads kommen.<\/p>\n\n\n\n
Unterschiede zu anderen Queues:<\/p>\n\n\n\n