java中等待所有线程都执行结束(转)

https://www.cnblogs.com/zl1991/p/6930160.html

 

转自:http://blog.csdn.net/liweisnake/article/details/12966761 

 

今天看到一篇文章,是关于java中如何等待所有线程都执行结束,文章总结得很好,原文如下http://software.intel.com/zh-cn/blogs/2013/10/15/java-countdownlatchcyclicbarrier/?utm_campaign=CSDN&utm_source=intel.csdn.net&utm_medium=Link&utm_content=others-%20Java

看过之后在想java中有很大的灵活性,应该有更多的方式可以做这件事。

这个事情的场景是这样的:许多线程并行的计算一堆问题,然后每个计算存在一个队列,在主线程要等待所有计算结果完成后排序并展示出来。这样的问题其实很常见。

1. 使用join。这种方式其实并不是那么的优雅,将所有线程启动完之后还需要将所有线程都join,但是每次join都会阻塞,直到被join线程完成,很可能所有被阻塞线程已经完事了,主线程还在不断地join,貌似有点浪费,而且两个循环也不太好看。

  1. 复制代码
     1 public void testThreadSync1() {  
     2   
     3     final Vector<Integer> list = new Vector<Integer>();  
     4     Thread[] threads = new Thread[TEST_THREAD_COUNT];  
     5     try {  
     6         for (int i = 0; i < TEST_THREAD_COUNT; i++) {  
     7             final int num = i;  
     8             threads[i] = new Thread(new Runnable() {  
     9                 public void run() {  
    10                     try {  
    11                         Thread.sleep(random.nextInt(100));  
    12                     } catch (InterruptedException e) {  
    13                         e.printStackTrace();  
    14                     }  
    15                     list.add(num);  
    16                     System.out.print(num + " add.\t");  
    17                 }  
    18             });  
    19             threads[i].start();  
    20         }  
    21         for (int i = 0; i < threads.length; i++) {  
    22             threads[i].join();  
    23             System.out.print(i + " end.\t");  
    24         }  
    25     } catch (InterruptedException ie) {  
    26         ie.printStackTrace();  
    27     }  
    28     printSortedResult(list);  
    29 }
    复制代码

     

  1. 1 9 add.  7 add.  3 add.  5 add.  4 add.  1 add.  0 add.  0 end.  1 end.  8 add.  2 add.  2 end.  3 end.  4 end.  5 end.  6 add.  6 end.  7 end.  8 end.  9 end.    
    2 before sort  
    3 9   7   3   5   4   1   0   8   2   6     
    4 after sort  
    5 0   1   2   3   4   5   6   7   8   9

     

 

2. 使用wait/notifyAll,这个方式其实跟上面是类似的,只是比较底层些吧(join实际上也是wait)。

 

  1. 复制代码
     1 @Test  
     2 public void testThreadSync2() throws IOException, InterruptedException {  
     3     final Object waitObject = new Object();  
     4     final AtomicInteger count = new AtomicInteger(TEST_THREAD_COUNT);  
     5     final Vector<Integer> list = new Vector<Integer>();  
     6     Thread[] threads = new Thread[TEST_THREAD_COUNT];  
     7     for (int i = 0; i < TEST_THREAD_COUNT; i++) {  
     8         final int num = i;  
     9         threads[i] = new Thread(new Runnable() {  
    10             public void run() {  
    11                 try {  
    12                     Thread.sleep(random.nextInt(100));  
    13                 } catch (InterruptedException e) {  
    14                     e.printStackTrace();  
    15                 }  
    16                 list.add(num);  
    17                 System.out.print(num + " add.\t");  
    18                 synchronized (waitObject) {  
    19                     int cnt = count.decrementAndGet();  
    20                     if (cnt == 0) {  
    21                         waitObject.notifyAll();  
    22                     }  
    23                 }  
    24             }  
    25         });  
    26         threads[i].start();  
    27     }  
    28     synchronized (waitObject) {  
    29         while (count.get() != 0) {  
    30             waitObject.wait();  
    31         }  
    32     }  
    33     printSortedResult(list);  
    34 }
    复制代码

     

3. 使用CountDownLatch,这其实是最优雅的写法了,每个线程完成后都去将计数器减一,最后完成时再来唤醒。

例1

  1. 复制代码
     1 @Test  
     2 public void testThreadSync3() {  
     3     final Vector<Integer> list = new Vector<Integer>();  
     4     Thread[] threads = new Thread[TEST_THREAD_COUNT];  
     5     final CountDownLatch latch = new CountDownLatch(TEST_THREAD_COUNT);  
     6     for (int i = 0; i < TEST_THREAD_COUNT; i++) {  
     7         final int num = i;  
     8         threads[i] = new Thread(new Runnable() {  
     9             public void run() {  
    10                 try {  
    11                     Thread.sleep(random.nextInt(100));  
    12                 } catch (InterruptedException e) {  
    13                     e.printStackTrace();  
    14                 }  
    15                 list.add(num);  
    16                 System.out.print(num + " add.\t");  
    17                 latch.countDown();  
    18             }  
    19         });  
    20         threads[i].start();  
    21     }  
    22     try {  
    23         latch.await();  
    24     } catch (InterruptedException e) {  
    25         e.printStackTrace();  
    26     }  
    27     printSortedResult(list);  
    28 }
    复制代码

     

例2

CountDownLatch 初始化设置count,即等待(await)count个线程或一个线程count次计数,通过工作线程来countDown计数减一,直到计数为0,await阻塞结束。

设置的count不可更改,如需要动态设置计数的线程数,可以使用CyclicBarrier.

 

下面的例子,所有的工作线程中准备就绪以后,并不是直接运行,而是等待主线程的信号后再执行具体的操作。

  1. 复制代码
      1 package com.example.multithread;  
      2   
      3 import java.util.concurrent.CountDownLatch;  
      4   
      5 class Driver  
      6 {  
      7     private static final int TOTAL_THREADS = 10;  
      8     private final CountDownLatch mStartSignal = new CountDownLatch(1);  
      9     private final CountDownLatch mDoneSignal = new CountDownLatch(TOTAL_THREADS);  
     10   
     11     void main()  
     12     {  
     13         for (int i = 0; i < TOTAL_THREADS; i++)  
     14         {  
     15             new Thread(new Worker(mStartSignal, mDoneSignal, i)).start();  
     16         }  
     17         System.out.println("Main Thread Now:" + System.currentTimeMillis());  
     18         doPrepareWork();// 准备工作   
     19         mStartSignal.countDown();// 计数减一为0,工作线程真正启动具体操作   
     20         doSomethingElse();//做点自己的事情   
     21         try  
     22         {  
     23             mDoneSignal.await();// 等待所有工作线程结束   
     24         }  
     25         catch (InterruptedException e)  
     26         {  
     27             // TODO Auto-generated catch block   
     28             e.printStackTrace();  
     29         }  
     30         System.out.println("All workers have finished now.");  
     31         System.out.println("Main Thread Now:" + System.currentTimeMillis());  
     32     }  
     33   
     34     void doPrepareWork()  
     35     {  
     36         System.out.println("Ready,GO!");  
     37     }  
     38   
     39     void doSomethingElse()  
     40     {  
     41         for (int i = 0; i < 100000; i++)  
     42         {  
     43             ;// delay   
     44         }  
     45         System.out.println("Main Thread Do something else.");  
     46     }  
     47 }  
     48   
     49 class Worker implements Runnable  
     50 {  
     51     private final CountDownLatch mStartSignal;  
     52     private final CountDownLatch mDoneSignal;  
     53     private final int mThreadIndex;  
     54   
     55     Worker(final CountDownLatch startSignal, final CountDownLatch doneSignal,  
     56             final int threadIndex)  
     57     {  
     58         this.mDoneSignal = doneSignal;  
     59         this.mStartSignal = startSignal;  
     60         this.mThreadIndex = threadIndex;  
     61     }  
     62   
     63     @Override  
     64     public void run()  
     65     {  
     66         // TODO Auto-generated method stub   
     67         try  
     68         {  
     69             mStartSignal.await();// 阻塞,等待mStartSignal计数为0运行后面的代码   
     70                                     // 所有的工作线程都在等待同一个启动的命令   
     71             doWork();// 具体操作   
     72             System.out.println("Thread " + mThreadIndex + " Done Now:"  
     73                     + System.currentTimeMillis());  
     74             mDoneSignal.countDown();// 完成以后计数减一   
     75         }  
     76         catch (InterruptedException e)  
     77         {  
     78             // TODO Auto-generated catch block   
     79             e.printStackTrace();  
     80         }  
     81     }  
     82   
     83     public void doWork()  
     84     {  
     85         for (int i = 0; i < 1000000; i++)  
     86         {  
     87             ;// 耗时操作   
     88         }  
     89         System.out.println("Thread " + mThreadIndex + ":do work");  
     90     }  
     91 }  
     92   
     93 public class CountDownLatchTest  
     94 {  
     95     public static void main(String[] args)  
     96     {  
     97         // TODO Auto-generated method stub   
     98         new Driver().main();  
     99     }  
    100   
    101 }
    复制代码

    通过Executor启动线程:

    1. 复制代码
       1 class CountDownLatchDriver2  
       2 {  
       3     private static final int TOTAL_THREADS = 10;  
       4     private final CountDownLatch mDoneSignal = new CountDownLatch(TOTAL_THREADS);  
       5  
       6 
       7   
       8     void main()  
       9     {  
      10         System.out.println("Main Thread Now:" + System.currentTimeMillis());  
      11         doPrepareWork();// 准备工作   
      12   
      13         Executor executor = Executors.newFixedThreadPool(TOTAL_THREADS);  
      14         for (int i = 0; i < TOTAL_THREADS; i++)  
      15         {  
      16             // 通过内建的线程池维护创建的线程   
      17             executor.execute(new RunnableWorker(mDoneSignal, i));  
      18         }  
      19         doSomethingElse();// 做点自己的事情   
      20         try  
      21         {  
      22             mDoneSignal.await();// 等待所有工作线程结束   
      23         }  
      24         catch (InterruptedException e)  
      25         {  
      26             // TODO Auto-generated catch block   
      27             e.printStackTrace();  
      28         }  
      29         System.out.println("All workers have finished now.");  
      30         System.out.println("Main Thread Now:" + System.currentTimeMillis());  
      31     }  
      32   
      33     void doPrepareWork()  
      34     {  
      35         System.out.println("Ready,GO!");  
      36     }  
      37   
      38     void doSomethingElse()  
      39     {  
      40         for (int i = 0; i < 100000; i++)  
      41         {  
      42             ;// delay   
      43         }  
      44         System.out.println("Main Thread Do something else.");  
      45     }  
      46 }  
      47   
      48 class RunnableWorker implements Runnable  
      49 {  
      50   
      51     private final CountDownLatch mDoneSignal;  
      52     private final int mThreadIndex;  
      53   
      54     RunnableWorker(final CountDownLatch doneSignal, final int threadIndex)  
      55     {  
      56         this.mDoneSignal = doneSignal;  
      57         this.mThreadIndex = threadIndex;  
      58     }  
      59   
      60     @Override  
      61     public void run()  
      62     {  
      63         // TODO Auto-generated method stub   
      64   
      65         doWork();// 具体操作   
      66         System.out.println("Thread " + mThreadIndex + " Done Now:"  
      67                 + System.currentTimeMillis());  
      68         mDoneSignal.countDown();// 完成以后计数减一   
      69                                 // 计数为0时,主线程接触阻塞,继续执行其他任务   
      70         try  
      71         {  
      72             // 可以继续做点其他的事情,与主线程无关了   
      73             Thread.sleep(5000);  
      74             System.out.println("Thread " + mThreadIndex  
      75                     + " Do something else after notifing main thread");  
      76   
      77         }  
      78         catch (InterruptedException e)  
      79         {  
      80             // TODO Auto-generated catch block   
      81             e.printStackTrace();  
      82         }  
      83   
      84     }  
      85   
      86     public void doWork()  
      87     {  
      88         for (int i = 0; i < 1000000; i++)  
      89         {  
      90             ;// 耗时操作   
      91         }  
      92         System.out.println("Thread " + mThreadIndex + ":do work");  
      93     }  
      94 }
      复制代码

      输出:

      复制代码
       1 Main Thread Now:1359959480786
       2 Ready,GO!
       3 Thread 0:do work
       4 Thread 0 Done Now:1359959480808
       5 Thread 1:do work
       6 Thread 1 Done Now:1359959480811
       7 Thread 2:do work
       8 Thread 2 Done Now:1359959480813
       9 Main Thread Do something else.
      10 Thread 3:do work
      11 Thread 3 Done Now:1359959480825
      12 Thread 5:do work
      13 Thread 5 Done Now:1359959480827
      14 Thread 7:do work
      15 Thread 7 Done Now:1359959480829
      16 Thread 9:do work
      17 Thread 9 Done Now:1359959480831
      18 Thread 4:do work
      19 Thread 4 Done Now:1359959480833
      20 Thread 6:do work
      21 Thread 6 Done Now:1359959480835
      22 Thread 8:do work
      23 Thread 8 Done Now:1359959480837
      24 All workers have finished now.
      25 Main Thread Now:1359959480838
      26 Thread 0 Do something else after notifing main thread
      27 Thread 1 Do something else after notifing main thread
      28 Thread 2 Do something else after notifing main thread
      29 Thread 3 Do something else after notifing main thread
      30 Thread 9 Do something else after notifing main thread
      31 Thread 7 Do something else after notifing main thread
      32 Thread 5 Do something else after notifing main thread
      33 Thread 4 Do something else after notifing main thread
      34 Thread 6 Do something else after notifing main thread
      35 Thread 8 Do something else after notifing main thread
      复制代码

       

 

4. 使用CyclicBarrier。这里其实类似上面,这个berrier只是在等待完成后自动调用传入CyclicBarrier的Runnable。

例1

  1. 复制代码
     1 @Test  
     2 public void testThreadSync4() throws IOException {  
     3     final Vector<Integer> list = new Vector<Integer>();  
     4     Thread[] threads = new Thread[TEST_THREAD_COUNT];  
     5     final CyclicBarrier barrier = new CyclicBarrier(TEST_THREAD_COUNT,  
     6             new Runnable() {  
     7                 public void run() {  
     8                     printSortedResult(list);  
     9                 }  
    10             });  
    11     for (int i = 0; i < TEST_THREAD_COUNT; i++) {  
    12         final int num = i;  
    13         threads[i] = new Thread(new Runnable() {  
    14             public void run() {  
    15                 try {  
    16                     Thread.sleep(random.nextInt(100));  
    17                 } catch (InterruptedException e) {  
    18                     e.printStackTrace();  
    19                 }  
    20                 list.add(num);  
    21                 System.out.print(num + " add.\t");  
    22                 try {  
    23                     barrier.await();  
    24                 } catch (InterruptedException e) {  
    25                     e.printStackTrace();  
    26                 } catch (BrokenBarrierException e) {  
    27                     e.printStackTrace();  
    28                 }  
    29             }  
    30         });  
    31         threads[i].start();  
    32     }  
    33     System.in.read();  
    34 }
    复制代码

    例2

    1. 复制代码
       1 class WalkTarget  
       2 {  
       3     private final int mCount = 5;  
       4     private final CyclicBarrier mBarrier;  
       5     ExecutorService mExecutor;  
       6   
       7     class BarrierAction implements Runnable  
       8     {  
       9         @Override  
      10         public void run()  
      11         {  
      12             // TODO Auto-generated method stub   
      13             System.out.println("所有线程都已经完成任务,计数达到预设值");  
      14             //mBarrier.reset();//恢复到初始化状态          
      15               
      16         }  
      17     }  
      18   
      19     WalkTarget()  
      20     {  
      21         //初始化CyclicBarrier   
      22         mBarrier = new CyclicBarrier(mCount, new BarrierAction());  
      23         mExecutor = Executors.newFixedThreadPool(mCount);  
      24   
      25         for (int i = 0; i < mCount; i++)  
      26         {  
      27             //启动工作线程   
      28             mExecutor.execute(new Walker(mBarrier, i));  
      29         }  
      30     }  
      31 }  
      32   
      33 //工作线程   
      34 class Walker implements Runnable  
      35 {  
      36     private final CyclicBarrier mBarrier;  
      37     private final int mThreadIndex;  
      38   
      39     Walker(final CyclicBarrier barrier, final int threadIndex)  
      40  
      41 
      42     {  
      43         mBarrier = barrier;  
      44         mThreadIndex = threadIndex;  
      45     }  
      46   
      47     @Override  
      48     public void run()  
      49     {  
      50         // TODO Auto-generated method stub   
      51         System.out.println("Thread " + mThreadIndex + " is running...");  
      52         // 执行任务   
      53         try  
      54         {  
      55             TimeUnit.MILLISECONDS.sleep(5000);  
      56             // do task   
      57         }  
      58         catch (InterruptedException e)  
      59         {  
      60             // TODO Auto-generated catch block   
      61             e.printStackTrace();  
      62         }  
      63   
      64         // 完成任务以后,等待其他线程完成任务   
      65         try  
      66         {  
      67             mBarrier.await();  
      68         }  
      69         catch (InterruptedException e)  
      70         {  
      71             // TODO Auto-generated catch block   
      72             e.printStackTrace();  
      73         }  
      74         catch (BrokenBarrierException e)  
      75         {  
      76             // TODO Auto-generated catch block   
      77             e.printStackTrace();  
      78         }  
      79         // 其他线程任务都完成以后,阻塞解除,可以继续接下来的任务   
      80         System.out.println("Thread " + mThreadIndex + " do something else");  
      81     }  
      82   
      83 }  
      84   
      85 public class CountDownLatchTest  
      86 {  
      87     public static void main(String[] args)  
      88     {  
      89         // TODO Auto-generated method stub   
      90         //new CountDownLatchDriver2().main();   
      91         new WalkTarget();  
      92     }  
      93   
      94 }
      复制代码

      输出(注意,只有所有的线程barrier.await之后才能继续执行其他的操作):

      Thread 0 is running... Thread 2 is running... Thread 3 is running... Thread 1 is running... Thread 4 is running... 所有线程都已经完成任务,计数达到预设值 Thread 4 do something else Thread 0 do something else Thread 2 do something else Thread 3 do something else Thread 1 do something else

 

5、

CountDownLatch和CyclicBarrier简单比较:

CountDownLatch

CyclicBarrier

软件包 java.util.concurrent java.util.concurrent
适用情景 主线程等待多个工作线程结束 多个线程之间互相等待,直到所有线程达到一个障碍点(Barrier point)
主要方法 CountDownLatch(int count) (主线程调用)

初始化计数

CountDownLatch.await (主线程调用)

阻塞,直到等待计数为0解除阻塞

CountDownLatch.countDown

计数减一(工作线程调用)

CyclicBarrier(int parties, Runnable barrierAction) //初始化参与者数量和障碍点执行Action,Action可选。由主线程初始化

CyclicBarrier.await() //由参与者调用

阻塞,直到所有线程达到屏障点

等待结束 各线程之间不再互相影响,可以继续做自己的事情。不再执行下一个目标工作。 在屏障点达到后,允许所有线程继续执行,达到下一个目标。可以重复使用CyclicBarrier
异常 如果其中一个线程由于中断,错误,或超时导致永久离开屏障点,其他线程也将抛出异常。
其他 如果BarrierAction不依赖于任何Party中的所有线程,那么在任何party中的一个线程被释放的时候,可以直接运行这个Action。

If(barrier.await()==2)

{

//do action

}

Leave a Reply

Your email address will not be published. Required fields are marked *