Это очень удобный класс для достижения синхронизации между двумя или более потоками, где один или несколько потоков могут ожидать завершения набора операций, выполняемых в других потоках (см. Javadoc и этот пост ). CountDownLatch может сэкономить ваше время в подходящих случаях, и вы должны знать об этом классе.
Как всегда, синхронизация потоков может вызвать взаимные блокировки, если код не очень хорош. И поскольку варианты использования параллелизма могут быть очень сложными, разработчики должны быть очень осторожны. Я не буду описывать здесь сложную проблему параллелизма, но простую проблему, с которой вы можете столкнуться при небрежном использовании CountDownLatch .
Предположим, у вас есть 2 потока (Thread-1 и Thread-2), которые совместно используют один java.util.concurrent.ArrayBlockingQueue, и вы хотите синхронизировать их с помощью CountDownLatch . Проверьте этот простой пример:
|
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
|
package gr.qiozas.simple.threads.countdownlatch; import java.util.concurrent.ArrayBlockingQueue;import java.util.concurrent.CountDownLatch; public class DeadlockCaseCDL { public static void main(String[] args) throws InterruptedException { CountDownLatch c = new CountDownLatch(1); ArrayBlockingQueue b = new ArrayBlockingQueue(1); new Thread(new T1(c, b)).start(); new Thread(new T2(c, b)).start(); } private static class T1 implements Runnable { private CountDownLatch c; private ArrayBlockingQueue b; private T1(CountDownLatch c, ArrayBlockingQueue b) { this.c = c; this.b = b; } public void run() { try { b.put(234); b.put(654); doWork(T1.class); c.countDown(); doWork(T1.class); } catch (InterruptedException ex) {} } } private static class T2 implements Runnable { private CountDownLatch c; private ArrayBlockingQueue b; private T2(CountDownLatch c, ArrayBlockingQueue b) { this.c = c; this.b = b; } public void run() { try { doWork(T2.class); c.await(); doWork(T2.class); System.out.println("I just dequeue "+b.take()); System.out.println("I just dequeue "+b.take()); } catch (InterruptedException ex) {} } } private static void doWork(Class classz) { System.out.println(classz.getName()+" do the work"); }} |
Вы видите выше, что основной поток создает CountDownLatch с количеством «1»? и ArrayBlockingQueue с емкостью «1? и после этого порождает «2 темы». ArrayBlockingQueue будет использоваться для реальной «работы» (постановка в очередь и удаление из очереди), а CountDownLatch будет использоваться для синхронизации потоков (постановка в очередь должна быть выполнена до удаления из очереди).
Поток-1 ставит в очередь 2 сообщения, а Поток-2 хочет удалить их из очереди, но только после того, как Поток-1 поставил в очередь оба сообщения. Эта синхронизация гарантируется CountDownLatch . Верите ли вы, что этот код в порядке ?? Нет, это не вызывает тупик !!!
Если вы внимательно посмотрите строку 10, вы увидите, что я инициализирую ArrayBlockingQueue с емкостью, равной «1?». Но Thread-1 хочет поставить в очередь 2 сообщения, а затем снять блокировку (CountDownLatch), чтобы впоследствии использовать ее для Thread-2. Вместимость «1? очереди блоков Thread-1, пока другой поток не выведет из очереди одно сообщение из очереди, а затем снова попытается поставить в очередь второе сообщение. К сожалению, только Thread-2 удаляет сообщения из очереди, но поскольку Thread-1 удерживает блокировку CountDownLatch, Thread-2 не может удалить любое сообщение и поэтому блокируется. Таким образом, у нас действительно есть тупик, поскольку оба потока заблокированы (ожидая получения разных блокировок). Thread-1 ожидает блокировки ArrayBlockingQueue и Thread-2 для блокировки CountDownLatch (вы можете увидеть это также в соответствующем дампе потока ниже).
Если мы увеличим емкость очереди, то этот код будет работать без проблем, но суть этой статьи не в этом. Вы должны понимать, что CountDownLatch следует использовать с осторожностью, чтобы избежать таких опасных случаев. Вы должны знать функциональные случаи вашего класса, разрабатывать эту функциональность для других разработчиков команды, писать полезный javadoc и всегда писать код, который является надежным в крайних случаях, а не только для удачных путей.
Еще один момент, который может вам помочь, заключается в том, что этот тупик не обнаруживается современными JVM. Попробуй.
Как вы, возможно, знаете, современные JVM (как Hotspot, так и JRockit) способны обнаруживать простые тупики и сообщать о них в Thread Dump. Посмотрите на простой пример взаимоблокировки, обнаруженный в JVM Hotspot:
Found one Java-level deadlock: ============================= "Thread-6": waiting to lock monitor 0x00a891ec (object 0x06c616e0, a java.lang.String), which is held by "Thread-9" "Thread-9": waiting to lock monitor 0x00a8950c (object 0x06c61708, a java.lang.String), which is held by "Thread-6" Вы можете попробовать DeadlockCaseCDL и получить дамп потока (в GNU / Linux запустите просто «kill -3‹ jvm_pid ›»). Вы увидите, что дамп потока выглядит нормально, и JVM не обнаруживает тупиков, но вы находитесь в тупике !!! Итак, имейте в виду, что этот вид тупика не обнаруживается JVM.
Проверьте этот пример дампа темы из моего локального исполнения:
Full thread dump Java HotSpot(TM) Server VM (17.1-b03 mixed mode): "DestroyJavaVM" prio=10 tid=0x0946e800 nid=0x5382 waiting on condition [0x00000000] java.lang.Thread.State: RUNNABLE "Thread-1" prio=10 tid=0x094b1400 nid=0x5393 waiting on condition [0x7c79a000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native Method) - parking to wait for (a java.util.concurrent.CountDownLatch$Sync) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:158) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:811) at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:969) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1281) at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:207) at gr.qiozas.simple.threads.countdownlatch.DeadlockCaseCDL$T2.run(DeadlockCaseCDL.java:50) at java.lang.Thread.run(Thread.java:662) "Thread-0" prio=10 tid=0x094afc00 nid=0x5392 waiting on condition [0x7c7eb000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native Method) - parking to wait for (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:158) at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1987) at java.util.concurrent.ArrayBlockingQueue.put(ArrayBlockingQueue.java:252) at gr.qiozas.simple.threads.countdownlatch.DeadlockCaseCDL$T1.run(DeadlockCaseCDL.java:29) at java.lang.Thread.run(Thread.java:662) "Low Memory Detector" daemon prio=10 tid=0x0947f800 nid=0x5390 runnable [0x00000000] java.lang.Thread.State: RUNNABLE "CompilerThread1" daemon prio=10 tid=0x7cfa9000 nid=0x538f waiting on condition [0x00000000] java.lang.Thread.State: RUNNABLE "CompilerThread0" daemon prio=10 tid=0x7cfa7000 nid=0x538e waiting on condition [0x00000000] java.lang.Thread.State: RUNNABLE "Signal Dispatcher" daemon prio=10 tid=0x7cfa5800 nid=0x538d waiting on condition [0x00000000] java.lang.Thread.State: RUNNABLE "Finalizer" daemon prio=10 tid=0x7cf96000 nid=0x538c in Object.wait() [0x7cd15000] java.lang.Thread.State: WAITING (on object monitor) at java.lang.Object.wait(Native Method) - waiting on (a java.lang.ref.ReferenceQueue$Lock) at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:118) - locked (a java.lang.ref.ReferenceQueue$Lock) at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:134) at java.lang.ref.Finalizer$FinalizerThread.run(Finalizer.java:159) "Reference Handler" daemon prio=10 tid=0x7cf94800 nid=0x538b in Object.wait() [0x7cd66000] java.lang.Thread.State: WAITING (on object monitor) at java.lang.Object.wait(Native Method) - waiting on (a java.lang.ref.Reference$Lock) at java.lang.Object.wait(Object.java:485) at java.lang.ref.Reference$ReferenceHandler.run(Reference.java:116) - locked (a java.lang.ref.Reference$Lock) "VM Thread" prio=10 tid=0x7cf92000 nid=0x538a runnable "GC task thread#0 (ParallelGC)" prio=10 tid=0x09475c00 nid=0x5383 runnable "GC task thread#1 (ParallelGC)" prio=10 tid=0x09477000 nid=0x5384 runnable "GC task thread#2 (ParallelGC)" prio=10 tid=0x09478800 nid=0x5385 runnable "GC task thread#3 (ParallelGC)" prio=10 tid=0x0947a000 nid=0x5387 runnable "VM Periodic Task Thread" prio=10 tid=0x09489800 nid=0x5391 waiting on condition JNI global references: 854 Heap PSYoungGen total 14976K, used 1029K [0xa2dd0000, 0xa3e80000, 0xb39d0000) eden space 12864K, 8% used [0xa2dd0000,0xa2ed1530,0xa3a60000) from space 2112K, 0% used [0xa3c70000,0xa3c70000,0xa3e80000) to space 2112K, 0% used [0xa3a60000,0xa3a60000,0xa3c70000) PSOldGen total 34304K, used 0K [0x815d0000, 0x83750000, 0xa2dd0000) object space 34304K, 0% used [0x815d0000,0x815d0000,0x83750000) PSPermGen total 16384K, used 1739K [0x7d5d0000, 0x7e5d0000, 0x815d0000) object space 16384K, 10% used [0x7d5d0000,0x7d782e90,0x7e5d0000)Full thread dump Java HotSpot(TM) Server VM (17.1-b03 mixed mode): "DestroyJavaVM" prio=10 tid=0x0946e800 nid=0x5382 waiting on condition [0x00000000] java.lang.Thread.State: RUNNABLE "Thread-1" prio=10 tid=0x094b1400 nid=0x5393 waiting on condition [0x7c79a000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native Method) - parking to wait for (a java.util.concurrent.CountDownLatch$Sync) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:158) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:811) at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:969) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1281) at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:207) at gr.qiozas.simple.threads.countdownlatch.DeadlockCaseCDL$T2.run(DeadlockCaseCDL.java:50) at java.lang.Thread.run(Thread.java:662) "Thread-0" prio=10 tid=0x094afc00 nid=0x5392 waiting on condition [0x7c7eb000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native Method) - parking to wait for (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:158) at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1987) at java.util.concurrent.ArrayBlockingQueue.put(ArrayBlockingQueue.java:252) at gr.qiozas.simple.threads.countdownlatch.DeadlockCaseCDL$T1.run(DeadlockCaseCDL.java:29) at java.lang.Thread.run(Thread.java:662) "Low Memory Detector" daemon prio=10 tid=0x0947f800 nid=0x5390 runnable [0x00000000] java.lang.Thread.State: RUNNABLE "CompilerThread1" daemon prio=10 tid=0x7cfa9000 nid=0x538f waiting on condition [0x00000000] java.lang.Thread.State: RUNNABLE "CompilerThread0" daemon prio=10 tid=0x7cfa7000 nid=0x538e waiting on condition [0x00000000] java.lang.Thread.State: RUNNABLE "Signal Dispatcher" daemon prio=10 tid=0x7cfa5800 nid=0x538d waiting on condition [0x00000000] java.lang.Thread.State: RUNNABLE "Finalizer" daemon prio=10 tid=0x7cf96000 nid=0x538c in Object.wait() [0x7cd15000] java.lang.Thread.State: WAITING (on object monitor) at java.lang.Object.wait(Native Method) - waiting on (a java.lang.ref.ReferenceQueue$Lock) at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:118) - locked (a java.lang.ref.ReferenceQueue$Lock) at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:134) at java.lang.ref.Finalizer$FinalizerThread.run(Finalizer.java:159) "Reference Handler" daemon prio=10 tid=0x7cf94800 nid=0x538b in Object.wait() [0x7cd66000] java.lang.Thread.State: WAITING (on object monitor) at java.lang.Object.wait(Native Method) - waiting on (a java.lang.ref.Reference$Lock) at java.lang.Object.wait(Object.java:485) at java.lang.ref.Reference$ReferenceHandler.run(Reference.java:116) - locked (a java.lang.ref.Reference$Lock) "VM Thread" prio=10 tid=0x7cf92000 nid=0x538a runnable "GC task thread#0 (ParallelGC)" prio=10 tid=0x09475c00 nid=0x5383 runnable "GC task thread#1 (ParallelGC)" prio=10 tid=0x09477000 nid=0x5384 runnable "GC task thread#2 (ParallelGC)" prio=10 tid=0x09478800 nid=0x5385 runnable "GC task thread#3 (ParallelGC)" prio=10 tid=0x0947a000 nid=0x5387 runnable "VM Periodic Task Thread" prio=10 tid=0x09489800 nid=0x5391 waiting on condition JNI global references: 854 Heap PSYoungGen total 14976K, used 1029K [0xa2dd0000, 0xa3e80000, 0xb39d0000) eden space 12864K, 8% used [0xa2dd0000,0xa2ed1530,0xa3a60000) from space 2112K, 0% used [0xa3c70000,0xa3c70000,0xa3e80000) to space 2112K, 0% used [0xa3a60000,0xa3a60000,0xa3c70000) PSOldGen total 34304K, used 0K [0x815d0000, 0x83750000, 0xa2dd0000) object space 34304K, 0% used [0x815d0000,0x815d0000,0x83750000) PSPermGen total 16384K, used 1739K [0x7d5d0000, 0x7e5d0000, 0x815d0000) object space 16384K, 10% used [0x7d5d0000,0x7d782e90,0x7e5d0000)
Ссылка: Выгодный CountDownLatch и хитрый java тупик от нашего партнера JCG Адрианоса Дадиса в блоге «Java, интеграция и достоинства исходного кода» .