{"id":29073,"date":"2022-04-20T17:07:46","date_gmt":"2022-04-20T15:07:46","guid":{"rendered":"https:\/\/www.happycoders.eu\/?p=29073"},"modified":"2024-11-27T15:08:38","modified_gmt":"2024-11-27T14:08:38","slug":"arrayblockingqueue-java","status":"publish","type":"post","link":"https:\/\/www.happycoders.eu\/de\/algorithmen\/arrayblockingqueue-java\/","title":{"rendered":"ArrayBlockingQueue in Java (mit Beispiel)"},"content":{"rendered":"\n

Im diesem Artikel geht es um die ArrayBlockingQueue<\/code> und deren Eigenschaften. Du siehst anhand eines Beispiels, wie die ArrayBlockingQueue<\/code> eingesetzt wird. Au\u00dferdem gebe ich dir eine Empfehlung, in welchen F\u00e4llen du diese Queue einsetzen solltest.<\/p>\n\n\n\n

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

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

ArrayBlockingQueue Eigenschaften<\/h2>\n\n\n\n

Die Klasse java.util.concurrent.ArrayBlockingQueue<\/code> basiert auf einem Array und ist \u2013 wie die meisten Queue-Implementierungen \u2013 threadsicher (s. u.<\/a>). Sie ist bounded (hat eine maximale Kapazit\u00e4t), entsprechend blockierend und bietet eine Fairness Policy (d. h. blockierende Methoden werden in der Reihenfolge bedient, in der sie aufgerufen wurden).<\/p>\n\n\n\n

Die Eigenschaften in der \u00dcbersicht:<\/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>
Array<\/td>Ja
(pessimistisches Locking mit einem Lock)<\/span><\/td>		Blocking<\/td>Optional<\/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

Aufgrund der m\u00f6glicherweise hohen Contention bei gleichzeitigem Schreib- und Lesezugriff solltest du \u2013 wenn du eine blockierende, threadsichere Queue ben\u00f6tigst \u2013 f\u00fcr deinen speziellen Einsatzweck testen, ob evtl. eine LinkedBlockingQueue<\/code> performanter ist. Diese basiert zwar auf einer verketteten Liste, verwendet allerdings zum Schreiben und Lesen zwei separate ReentrantLock<\/code>s, was die Zugangskonflikte reduziert.<\/p>\n\n\n\n

ArrayBlockingQueue Beispiel<\/h2>\n\n\n\n

Im folgenden Beispiel erzeugen wir eine ArrayBlockingQueue mit der Kapazit\u00e4t 3. Dann lassen wir \u00fcber einen ScheduledExecutorService<\/code> in bestimmten Abst\u00e4nden Elemente in die Queue schreiben und aus ihr lesen (\u2192 Code auf GitHub<\/a>):<\/p>\n\n\n

public<\/span> class<\/span> ArrayBlockingQueueExample<\/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> ArrayBlockingQueue<>(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
Wir versuchen alle drei Sekunden, beginnend sofort, ein Element aus der Queue zu lesen. Wir schreiben die Elemente sek\u00fcndlich, fangen allerdings erst nach 3,5 s damit an. Zu diesem Zeitpunkt sollten also bereits zwei lesende Threads blockiert haben und darauf warten, dass Elemente in die Queue geschrieben werden.<\/p>\n\n\n\n
Da wir schneller schreiben als lesen, sollte die Queue bald ihre Kapazit\u00e4tsgrenze erreicht haben. Ab dem Moment sollten die schreibenden Threads so lange blockieren, bis die lesenden Threads aufgeholt haben.<\/p>\n\n\n\n
Hier eine beispielhafte Ausgabe:<\/p>\n\n\n
[ 0.0s] [pool-1-thread-1 ]     Calling queue.take() (queue = [])...\n[ 3.0s] [pool-1-thread-2 ]     Calling queue.take() (queue = [])...\n[ 3.5s] [pool-1-thread-3 ] Calling queue.put(0) (queue = [])...\n[ 3.5s] [pool-1-thread-3 ] queue.put(0) returned (queue = [])\n[ 3.5s] [pool-1-thread-1 ]     queue.take() returned 0 (queue = [])\n[ 4.5s] [pool-1-thread-9 ] Calling queue.put(1) (queue = [])...\n[ 4.5s] [pool-1-thread-9 ] queue.put(1) returned (queue = [])\n[ 4.5s] [pool-1-thread-2 ]     queue.take() returned 1 (queue = [])\n[ 5.5s] [pool-1-thread-7 ] Calling queue.put(2) (queue = [])...\n[ 5.5s] [pool-1-thread-7 ] queue.put(2) returned (queue = [2])\n[ 6.0s] [pool-1-thread-8 ]     Calling queue.take() (queue = [2])...\n[ 6.0s] [pool-1-thread-8 ]     queue.take() returned 2 (queue = [])\n[ 6.5s] [pool-1-thread-5 ] Calling queue.put(3) (queue = [])...\n[ 6.5s] [pool-1-thread-5 ] queue.put(3) returned (queue = [3])\n[ 7.5s] [pool-1-thread-4 ] Calling queue.put(4) (queue = [3])...\n[ 7.5s] [pool-1-thread-4 ] queue.put(4) returned (queue = [3, 4])\n[ 8.5s] [pool-1-thread-10] Calling queue.put(5) (queue = [3, 4])...\n[ 8.5s] [pool-1-thread-10] queue.put(5) returned (queue = [3, 4, 5])\n[ 9.0s] [pool-1-thread-6 ]     Calling queue.take() (queue = [3, 4, 5])...\n[ 9.0s] [pool-1-thread-6 ]     queue.take() returned 3 (queue = [4, 5])\n[ 9.5s] [pool-1-thread-3 ] Calling queue.put(6) (queue = [4, 5])...\n[ 9.5s] [pool-1-thread-3 ] queue.put(6) returned (queue = [4, 5, 6])\n[10.5s] [pool-1-thread-1 ] Calling queue.put(7) (queue = [4, 5, 6])...\n[11.5s] [pool-1-thread-9 ] Calling queue.put(8) (queue = [4, 5, 6])...\n[12.0s] [pool-1-thread-2 ]     Calling queue.take() (queue = [4, 5, 6])...\n[12.0s] [pool-1-thread-2 ]     queue.take() returned 4 (queue = [5, 6, 7])\n[12.0s] [pool-1-thread-1 ] queue.put(7) returned (queue = [5, 6, 7])\n[12.5s] [pool-1-thread-7 ] Calling queue.put(9) (queue = [5, 6, 7])...\n[15.0s] [pool-1-thread-8 ]     Calling queue.take() (queue = [5, 6, 7])...\n[15.0s] [pool-1-thread-8 ]     queue.take() returned 5 (queue = [6, 7, 8])\n[15.0s] [pool-1-thread-9 ] queue.put(8) returned (queue = [6, 7, 8])\n[18.0s] [pool-1-thread-5 ]     Calling queue.take() (queue = [6, 7, 8])...\n[18.0s] [pool-1-thread-5 ]     queue.take() returned 6 (queue = [7, 8, 9])\n[18.0s] [pool-1-thread-7 ] queue.put(9) returned (queue = [7, 8, 9])\n[21.0s] [pool-1-thread-4 ]     Calling queue.take() (queue = [7, 8, 9])...\n[21.0s] [pool-1-thread-4 ]     queue.take() returned 7 (queue = [8, 9])\n[24.0s] [pool-1-thread-10]     Calling queue.take() (queue = [8, 9])...\n[24.0s] [pool-1-thread-10]     queue.take() returned 8 (queue = [9])\n[27.0s] [pool-1-thread-6 ]     Calling queue.take() (queue = [9])...\n[27.0s] [pool-1-thread-6 ]     queue.take() returned 9 (queue = [])<\/code><\/span>Code-Sprache:<\/span> Klartext<\/span> (<\/span>plaintext<\/span>)<\/span><\/small><\/pre>\n\n\n
Wie vorausgesehen blockieren die ersten zwei Leseversuche bei 0,0 s und 3,0 s, da noch keine Elemente in die Queue geschrieben wurden.<\/p>\n\n\n\n
Nach 3,5 s wird das erste Element geschrieben. Dadurch wird der erste Thread aufgeweckt und entnimmt dieses Element wieder. Nach 4,5 s wird das zweite Element geschrieben und der zweite Thread aufgeweckt, um das Element zu entnehmen.<\/p>\n\n\n\n
Da das Programm schneller schreibt als liest, blockieren nach 10,5 s Thread 1, nach 11,5 s Thread 9 und nach 12,5 s Thread 7 beim Versuch weitere Elemente in die zu dem Zeitpunkt volle Queue zu schreiben. <\/p>\n\n\n\n
Nach 12,0 s wird ein Element entnommen und Thread 1 kann mit dem Schreiben fortfahren. Nach 15,0 s wird ein weiteres Element entnommen und Thread 9 kann fortfahren. Nach 18,0 s kann Thread 7 fortfahren. <\/p>\n\n\n\n
Da keine weiteren Elemente in die Queue geschrieben werden, leert sie sich gegen Ende wieder.<\/p>\n\n\n\n
Ist ArrayBlockingQueue threadsicher?<\/h2>\n\n\n\n
Ja, ArrayBlockingQueue<\/code> ist threadsicher.<\/p>\n\n\n\n
Die Threadsicherheit von ArrayBlockingQueue<\/code> wird durch ein einzelnes ReentrantLock<\/code> gew\u00e4hrleistet. Dieses wird sowohl f\u00fcr den Kopf als auch f\u00fcr das Ende der Queue verwendet, so dass es bei gleichzeitigen Schreib- und Lesezugriffen zu Zugriffskonflikten (\"thread contention\") zwischen Producer- und Consumer-Threads kommen kann.<\/p>\n\n\n\n
Explizite Locks wie ReentrantLock<\/code> sind haupts\u00e4chlich f\u00fcr Einsatzgebiete geeignet, in denen es zu hoher Thread Contention kommt. Bei niedriger bis moderater Contention ist optimistisches Locking performanter.<\/p>\n\n\n\n
Unterschiede zu anderen Queues:<\/p>\n\n\n\n