首页 > 技术文章 > 并发工具类:CountDownLatch、CyclicBarrier、Semaphore

dongguacai 2016-11-02 16:40 原文

在多线程的场景下,有些并发流程需要人为来控制,在JDK的并发包里提供了几个并发工具类:CountDownLatch、CyclicBarrier、Semaphore。

一、CountDownLatch

 1 import java.util.concurrent.CountDownLatch;
 2 
 3 
 4 public class CountDownLatchTest
 5 {
    //设置N为2
6 static CountDownLatch c = new CountDownLatch(2); 7 public static void main(String[] args) throws Exception 8 { 9 Thread t1 = new Thread(new Runnable() 10 { 11 12 @Override 13 public void run() 14 { 15 System.out.println("1");
           //将N减1
16 c.countDown(); 17 } 18 }); 19 20 Thread t2 = new Thread(new Runnable() 21 { 22 23 @Override 24 public void run() 25 { 26 System.out.println("2"); 27 c.countDown(); 28 } 29 }); 30 t1.start(); 31 t2.start();
      //让当前线程等待,直到计数N为0
32 c.await(); 33 System.out.println("3"); 34 } 35 }

执行结果:

1
2
3

这里会存在两种结果:123或者213,但是绝对不会出现3打印在1、2前面的。

new CountDownLatch(2);

这个CountDownLatch的构造函数接收一个int类型的参数作为计数器,N表示阻塞的线程必须等待N次countDown才能执行。

每次调用CountDownLatch的countDown方法时,N就会减1,而这个方法可以使用在任何地方,这里的N点可以是N个线程,也可以是一个线程中N个步骤。

而CountDownLatch的await方法则会阻塞当前线程,直到N为0的时候才能执行。

我们将上面的程序改造下,让线程中有两个打印动作,并且第二个动作前线程休眠一段时间:

 1 import java.util.concurrent.CountDownLatch;
 2 
 3 
 4 public class CountDownLatchTest
 5 {
 6     static CountDownLatch c = new CountDownLatch(2);
 7     public static void main(String[] args) throws Exception
 8     {
 9         Thread t1 = new Thread(new Runnable()
10         {
11             
12             @Override
13             public void run()
14             {
15                 System.out.println("1");
16                 c.countDown();
17                 try
18                 {
19                     Thread.sleep(500);
20                     System.out.println("2");
21                 }
22                 catch (InterruptedException e)
23                 {
24                     e.printStackTrace();
25                 }
26             }
27         });
28         
29         Thread t2 = new Thread(new Runnable()
30         {
31             
32             @Override
33             public void run()
34             {
35                 System.out.println("3");
36                 c.countDown();
37             }
38         });
39         t1.start();
40         t2.start();
41         c.await();
42         System.out.println("4");
43     }
44 }

执行结果:

1
3
4
2

这个结果是由于在打印完1、3之后,N已经变化为0,主线程执行打印4,由于线程1休眠,所以2最后才打印。

在上面的程序中,如果将N设置为3,则主线程中的打印4永远不会执行,因为没有N永远只会到1而不会减少到0.

在这里我们想起了线程的join方法,这个方法也是可以阻塞当前线程,等待某线程执行完成。通过对比,我们可以发现使用CountDownLatch这个工具类更灵活,因为countDown可以用在任何线程的任何地方。

CountDownLatch适合一个大任务拆分成多个小任务,然后在所有子任务完成后,通知其他的后续操作开始执行。

二、同步屏障CyclicBarrier

CyclicBarrier默认的构造方法CyclicBarrier(int parties),其参数表示屏障拦截的线程数量,每个线程调用await方法告诉CyclicBarrier我已经到达屏障,然后当前线程被阻塞,直到被拦截的线程全部都到达了屏障,然后前面被阻塞的线程才能开始执行,否则会被一直阻塞。

 1 public class CyclicBarrierTest
 2 {
 3     static CyclicBarrier c = new CyclicBarrier(3);
 4     
 5     public static void main(String[] args)
 6         throws Throwable, BrokenBarrierException
 7     {
 8         Thread t1 = new Thread(new Runnable()
 9         {
10             @Override
11             public void run()
12             {
13                 try
14                 {
15                     c.await();
16                 }
17                 catch (InterruptedException e)
18                 {
19                     e.printStackTrace();
20                 }
21                 catch (BrokenBarrierException e)
22                 {
23                     e.printStackTrace();
24                 }
25                 System.out.println("1");
26                 
27             }
28         });
29         Thread t2 = new Thread(new Runnable()
30         {
31             @Override
32             public void run()
33             {
34                 try
35                 {
36                     c.await();
37                 }
38                 catch (InterruptedException e)
39                 {
40                     e.printStackTrace();
41                 }
42                 catch (BrokenBarrierException e)
43                 {
44                     e.printStackTrace();
45                 }
46                 System.out.println("2");
47             }
48         });
49         
50         t1.start();
51         t2.start();
52         c.await();
53         System.out.println("3");
54     }
55 }

执行结果:

3
1
2

上述中被屏障拦截的线程有3个,其中线程1和线程2执行的时候先到达屏障,然后被阻塞,主线程执行第52行到达屏障,至此阻塞的三个线程全部到达屏障,然后阻塞的线程可以去竞争CPU开始执行。

如果将拦截的线程数修改为4:

static CyclicBarrier c = new CyclicBarrier(4);

这样的话被拦截的线程数有4个,但是只有三个线程调用await方法告诉CyclicBarrier,我到达了屏障。所以这三个线程都会被阻塞。

另外还有一点就是CyclicBarrier的计数器可以重置,例如设置的是拦截线程数量为2,但是有3个线程调用了await()方法表示到达了屏障,这个时候会出现最先达到屏障的两个线程顺利执行完毕,而最后到达的第三个线程则一直被阻塞,因为它等不到另外一个线程到达屏障了。

而如果拦截线程的数量依旧为2,但是有4个线程调用了await()方法,那么这4个线程是分两批执行的,前两个线程满足拦截的线程数,到达屏障后放行;然后CyclicBarrier的计数器重置,后面两个线程到达屏障后放行。

 1 import java.util.concurrent.BrokenBarrierException;
 2 import java.util.concurrent.CyclicBarrier;
 3 
 4 import sun.java2d.SunGraphicsEnvironment.T1Filter;
 5 
 6 
 7 public class CyclicBarrierTest
 8 {
 9     static CyclicBarrier c = new CyclicBarrier(2);
10     
11     public static void main(String[] args) throws Throwable, BrokenBarrierException
12     {
13         Thread t1 = new Thread(new Runnable()
14         {
15             @Override
16             public void run()
17             {
18                 try
19                 {
20                     c.await();
21                 }
22                 catch (InterruptedException e)
23                 {
24                     e.printStackTrace();
25                 }
26                 catch (BrokenBarrierException e)
27                 {
28                     e.printStackTrace();
29                 }
30                 System.out.println("1");
31                 
32             }
33         });
34         Thread t2 = new Thread(new Runnable()
35         {
36             @Override
37             public void run()
38             {
39                 try
40                 {
41                     c.await();
42                 }
43                 catch (InterruptedException e)
44                 {
45                     e.printStackTrace();
46                 }
47                 catch (BrokenBarrierException e)
48                 {
49                     e.printStackTrace();
50                 }
51                 System.out.println("2");
52             }
53         });
54         
55         Thread t3 = new Thread(new Runnable()
56         {
57             @Override
58             public void run()
59             {
60                 try
61                 {
62                     c.await();
63                 }
64                 catch (InterruptedException e)
65                 {
66                     e.printStackTrace();
67                 }
68                 catch (BrokenBarrierException e)
69                 {
70                     e.printStackTrace();
71                 }
72                 System.out.println("3");
73                 
74             }
75         });
76         t1.start();
77         t2.start();
78         t3.start();
79         c.await();
80         System.out.println("4");
81     }
82 }

CyclicBarrier可以用于多线程计算数据,最后合并结果的场景;由于CyclicBarrier的计数器可以重置,所以可以使用它处理更为复杂的业务场景,而CountDownLatch计数器只能使用一次。

三、信号量Semaphore

无论是内部锁synchronized还是重入锁ReentrantLock,一次都只允许一个线程访问某一个资源,而信号量却可以指定多个线程同时访问某一资源;主要用来控制同时访问特定资源的线程数量,它通过协调各个线程,以保证合理的使用公共资源。

 1 public class SemaphoreTest implements Runnable
 2 {
 3     final Semaphore s = new Semaphore(5);
 4     @Override
 5     public void run()
 6     {
 7         try
 8         {
 9             s.acquire();
10             Thread.sleep(1000);
11             System.out.println(Thread.currentThread().getName() + " is done");
12             s.release();
13         }
14         catch (InterruptedException e)
15         {
16             e.printStackTrace();
17         }
18     }
19 }
public class MainTest
{
    public static void main(String[] args)
    {
        SemaphoreTest s = new SemaphoreTest();
        //创建一个可重用固定线程数的线程池,线程数量为20
        ExecutorService threadPool= Executors.newFixedThreadPool(20);
        for(int i=0;i<20;i++)
        {
            threadPool.submit(s);
        }
        threadPool.shutdown();
    }
}

执行的结果是每五个线程为一组打印消息。

线程池里面有20个可重复使用的线程数量,但是信号量只有5个,也就是每次只能并发5个线程执行,其他线程阻塞。

信号量为5,可以认为线程池里有5把锁,每个线程调用acquire和release分别表示获取锁和释放锁,这样,通过信号量就可以调度多个线程的执行。

 

推荐阅读