Статьи

Учебник по параллелизму Java — CountDownLatch

Май 17, 2019
Некоторым утилитам параллелизма в Java, естественно, уделяется больше внимания, чем другим только потому, что они служат задачам общего назначения, а не более конкретным. Большинство из нас довольно часто сталкиваются с такими вещами, как услуги исполнителя и одновременные коллекции.

Другие утилиты встречаются реже, поэтому иногда они могут уйти от нас, но о них стоит помнить. CountDownLatch является одним из таких инструментов.

CountDownLatch — более общий механизм ожидания / уведомления

Разработчики Java всех видов должны быть знакомы с подходом ожидания / уведомления к блокировке, пока не будет достигнуто условие. Вот небольшой пример того, как это работает:

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
public void testWaitNotify() throws Exception {
   final Object mutex = new Object();
   Thread t = new Thread() {
      public void run() {
      // we must acquire the lock before waiting to be notified
      synchronized(mutex) {
         System.out.println("Going to wait " +
                            "(lock held by " + Thread.currentThread().getName() + ")");
            try {
               mutex.wait(); // this will release the lock to be notified (optional timeout can be supplied)
            } catch (InterruptedException e) {
               e.printStackTrace();
            }
 
            System.out.println("Done waiting " +
                               "(lock held by " + Thread.currentThread().getName() + ")");
         }
      }
   };
 
   t.start(); // start her up and let her wait()
 
   // not normally how we do things, but good enough for demonstration purposes
   Thread.sleep(1000);
 
   // we acquire the lock released by wait(), and notify()
   synchronized (mutex) {
      System.out.println("Going to notify " +
                         "(lock held by " + Thread.currentThread().getName() + ")");
      mutex.notify();
      System.out.println("Done notify " +
                         "(lock held by " + Thread.currentThread().getName() + ")");
   }
 
}

Выход

1
2
3
4
Going to wait (lock held by Thread-0)
Going to notify (lock held by main)
Done notify (lock held by main)
Done waiting (lock held by Thread-0)

CountDownLatch может фактически использоваться подобно ожиданию / уведомлению только с одним уведомлением — то есть, если вы не хотите, чтобы wait () останавливалось, если notify () вызывается до того, как вы получили блокировку и вызвали wait () , Это на самом деле более снисходительно, а в некоторых случаях это именно то, что вы хотите. Вот пример:

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
public void testWaitNotify() throws Exception {
   final CountDownLatch latch = new CountDownLatch(1); // just one time
   Thread t = new Thread() {
      public void run() {
         // no lock to acquire!
         System.out.println("Going to count down...");
         latch.countDown();
      }
   };
 
   t.start(); // start her up and let her wait()
   System.out.println("Going to await...");
   latch.await();
   System.out.println("Done waiting!");
}

Как видите, это проще, чем ждать / уведомлять, и требует меньше кода. Это также позволяет нам вызывать условие, которое в конечном итоге освобождает блок перед вызовом wait () . Это может означать более безопасный код.

Пример из реального мира

Итак, мы знаем, что можем использовать его как более простой механизм ожидания / уведомления, но вы, вероятно, видели аргумент конструктора, который мы использовали выше. В конструкторе вы указываете, сколько раз защелка должна быть отсчитана перед разблокировкой. Какие возможные варианты использования есть для этого? Что ж, его можно использовать, чтобы заставить процесс ждать, пока не будет выполнено определенное количество действий.

Например, если у вас есть асинхронный процесс, который вы можете подключить через прослушиватели или что-то подобное, вы можете создать модульные тесты, которые проверят, было ли выполнено определенное количество вызовов. Это позволяет нам тратить только столько времени, сколько нам нужно в обычном случае (или некоторый предел, прежде чем мы сделаем залог и допустим неудачу).

Недавно я столкнулся со случаем, когда мне пришлось проверить, что сообщения JMS извлекаются из очереди и обрабатываются правильно. Это было естественно асинхронно и вне моего контроля, и макеты не были опцией, так как это было полностью собранное приложение с контекстом Spring и т. Д. Чтобы проверить это, я внес небольшие изменения в потребляющие сервисы, чтобы позволить слушателям вызываться, когда сообщения были обработаны. Затем я смог временно добавить слушателя, который использовал 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
public void testSomeProcessing() throws Exception {
   // should be called twice
   final CountDownLatch testLatch = new CountDownLatch(2);
   ExecutorService executor = Executors.newFixedThreadPool(1);
   AsyncProcessor processor = new AsyncProcessor(new Observer() {
      // this observer would be the analogue for a listener in your async process
      public void update(Observable o, Object arg) {
         System.out.println("Counting down...");
         testLatch.countDown();
      }
   });
 
   //submit two tasks to be process
   // (in my real world example, these were JMS messages)
   executor.submit(processor);
   executor.submit(processor);
 
   System.out.println("Submitted tasks. Time to wait...");
   long time = System.currentTimeMillis();
   testLatch.await(5000, TimeUnit.MILLISECONDS); // bail after a reasonable time
   long totalTime = System.currentTimeMillis() - time;
 
   System.out.println("I awaited for " + totalTime +
                      "ms. Did latch count down? " + (testLatch.getCount() == 0));
 
   executor.shutdown();
}
 
// just a process that takes a random amount of time
// (up to 2 seconds) and calls its listener
public class AsyncProcessor implements Callable<Object> {
      private Observer listener;
      private AsyncProcessor(Observer listener) {
      this.listener = listener;
   }
 
   public Object call() throws Exception {
      // some processing here which can take all kinds of time...
      int sleepTime = new Random().nextInt(2000);
      System.out.println("Sleeping for " + sleepTime + "ms");
      Thread.sleep(sleepTime);
      listener.update(null, null); // not standard usage, but good for a demo
      return null;
   }
}

Выход

1
2
3
4
5
6
Submitted tasks. Time to wait...
Sleeping for 739ms
Counting down...
Sleeping for 1742ms
Counting down...
I awaited for 2481ms. Did latch count down? true

Вывод

Вот и все об этом для CountDownLatch . Это не сложный предмет, и его использование ограничено, но приятно видеть примеры и знать, что они у вас в сундуке с инструментами, когда вы сталкиваетесь с проблемой, как я. В будущем я определенно буду помнить об этом для более простого ожидания / уведомления, если ничего больше. Если у вас есть вопросы или комментарии по поводу этого поста или других в серии, просто оставьте сообщение.

Ссылка: Java Concurrency, часть 6 — CountDownLatch от наших партнеров по JCG в блоге Carfey Software .

Статьи по Теме :
44
Share: