Java并发编程高级内容介绍

计数器：CountDownLatch

CountDownLatch类似于一个计数器，和Atomic类比较相近，操作是原子的，即多个线程同时只能有一个可以去操作。CountDownLatch对象设置一个初始的数字作为计数值，任何调用这个对象上的await()方法都会阻塞，直到这个计数器的计数值被其他的线程调用countDown()减为0为止。典型的应用场景就是：有一个任务想要往下执行，但必须要等到其他的任务执行完毕后才可以继续往下执行。例如在Zookeeper的使用过程中，由于客户端与服务器建立连接是异步调用的，因此主线程需要await()阻塞直至异步回调countDown()完成。

代码示例：

1 2 3 4 5 6 7 8 9 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 54 55 56 57

public class CountDownLatchTest {     public static void main(String[] args) {         final CountDownLatch countDownLatch = new CountDownLatch(2);         Thread work1 = new Thread(new Runnable() {             @Override             public void run() {                 System.out.println(Thread.currentThread() + " doing work...start");                 try {                     Thread.sleep(200);                 } catch (InterruptedException e) {                     e.printStackTrace();                 }                 System.out.println(Thread.currentThread() + " doing work...end ");                 countDownLatch.countDown();             }         },"work1");         Thread work2 = new Thread(new Runnable() {             @Override             public void run() {                 System.out.println(Thread.currentThread().getName() + " doing work...start");                 try {                     Thread.sleep(200);                 } catch (InterruptedException e) {                     e.printStackTrace();                 }                 System.out.println(Thread.currentThread().getName() + " doing work...end ");                 countDownLatch.countDown();             }         },"work2");         work1.start();         work2.start();         try {             countDownLatch.await();             System.out.println("all workers finish ");         } catch (InterruptedException e) {             e.printStackTrace();         }     } }

齐步走：CyclicBarrier

Barrier的意思是栅栏，就是让一组线程相互等待，直至所有线程都到齐了，那么就可以齐步走。Cyclic是循环的意思，就是说Barrier可以循环使用。CyclicBarrier主要的方法就是await()，较CountDownLatch的await()虽然都是阻塞，但是CyclicBarrier.await()有返回值int，即当前线程是第几个到达这个Barrier的线程。

构造CyclicBarrier时指定计数值，await() 方法每被调用一次，计数便会减少1，并阻塞住当前线程。当计数减至0时，阻塞解除，所有在此 CyclicBarrier 上面阻塞的线程开始运行。在这之后，如果再次调用 await() 方法，计数就又会变成 N-1，新一轮重新开始。在构造方法上还可以传递一个Runnable对象，阻塞解除时这个Runnable会得到运行。

CyclicBarrier有点“不见不散”的味道，想一想，如果某个成员因某种原因来不了Barrier这个地方，那么我们一直等待吗？实际中，如果来不了理应通知其他成员，别等了，回家吧！注意到CyclicBarrier.await()独有的BrokenBarrierException异常

代码示例：

1 2 3 4 5 6 7 8 9 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 54 55

public class CyclicBarrierTest {     public static void main(String[] args) {         final CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {             @Override             public void run() {                 System.out.println("都准备好啦!");             }         });         Thread runman1 = new Thread(new Runnable() {             @Override             public void run() {                 try {                     Thread.sleep(200);                 } catch (InterruptedException e) {                     e.printStackTrace();                 }                 try {                     cyclicBarrier.await();                     System.out.println(Thread.currentThread().getName() + "i am ok");                 } catch (InterruptedException e) {                     e.printStackTrace();                 } catch (BrokenBarrierException e) {                     e.printStackTrace();                 }             }         },"runman1");         Thread runman2 = new Thread(new Runnable() {             @Override             public void run() {                 try {                     cyclicBarrier.await();                     System.out.println(Thread.currentThread().getName() + "i am ok");                 } catch (InterruptedException e) {                     e.printStackTrace();                 } catch (BrokenBarrierException e) {                     e.printStackTrace();                 }             }         },"runman2");         runman1.start();         runman2.start();     } }

Callable And Future

在博主以前的博客《Java Future模式实现》中有介绍Future模式，Future模式非常适合在处理耗时很长的业务逻辑，可以有效的减少系统的响应时间，提高系统的吞吐量。JDK其实已经为我们提供了API实现，我们来看一段代码即可：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31

public class FutureTest {     public static void main(String[] args) {         FutureTask<String> futureTask = new FutureTask<String>(new Callable<String>() {             @Override             public String call() throws Exception {                 Thread.sleep(2000);                 return "ok";             }         });         ExecutorService es = Executors.newFixedThreadPool(1);         es.submit(futureTask);         System.out.println("开启线程去异步处理，主线程继续往下执行!");         try {             System.out.println("取得异步处理结果：" + futureTask.get());         } catch (InterruptedException e) {             e.printStackTrace();         } catch (ExecutionException e) {             e.printStackTrace();         }     } }

注意到线程池执行任务，可以利用2个方法：

submit和execute有什么区别呢？从入参和结果类型就知道了。

信号量：Semaphore

Semaphore实现的功能就类似厕所有5个坑，假如有10个人要上厕所，那么显然同时只能有5个人占用厕所，当5个人中的任何一个人让开后，其中等待的另外5个人中又有一个人可以占用了。另外等待的5个人中可以是随机获得优先机会，也可以是按照先来后到的顺序获得机会，这取决于构造Semaphore对象时传入的fair参数选项。

Semaphore可以控制某个资源可被同时访问的个数(构造方法传入)，通过 acquire() 获取一个许可，如果没有就等待，而 release() 释放一个许可。

代码示例：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33

public static void main(String[] args) {     final Semaphore semaphore = new Semaphore(5);     for(int i = 0 ; i < 6 ; i++){         new Thread(new Runnable() {             @Override             public void run() {                 try {                     semaphore.acquire();                     System.out.println(Thread.currentThread().getName() + " 运行...");                     Thread.sleep(1000);                     semaphore.release();                     System.out.println(Thread.currentThread().getName() + " 结束...");                 } catch (InterruptedException e) {                     e.printStackTrace();                 }             }         },String.valueOf(i)).start();     } }

Condition

JDK由原始的synchronized发展到Lock，以类的方式提供锁机制，发展出重入锁、读写锁，以类的形式存在自然功能更加强大灵活，比如可以tryLock进行锁的嗅探。在synchronized代码块中我们可以使用wait/notify/notifyAll来进行线程的协同工作，那么JDK也发展了这一块，即Condition。Condition.await类似于wait,Condition.signal/signalAll类似于notify/nofityAll。下面我们简单实现一个Condition版的生产者/消费者。

处理核心：Handler

1 2 3 4 5 6 7 8 9 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 54 55 56 57 58 59 60

public class Handler {     //容器     private LinkedList<String> linkedList = new LinkedList<String>();     //限制     private int MAX_SIZE = 3;     //锁     private Lock lock = new ReentrantLock();     //condition  实际上，可以new多个condition，这里暂且只是用给一个     private Condition condition = lock.newCondition();     public void put(String bread){         try{             lock.lock();             if(linkedList.size() == MAX_SIZE){                 System.out.println("容器已满");                 condition.await();             }             linkedList.add(bread);             System.out.println("放入面包" + bread);             condition.signalAll();         }catch (Exception e){             e.printStackTrace();         }finally {             lock.unlock();         }     }     public void eat(){         try{             lock.lock();             if(linkedList.size() == 0){                 System.out.println("容器为空");                 condition.await();             }             String bread = linkedList.removeFirst();             System.out.println("吃掉一个面包" + bread);             condition.signalAll();         }catch(Exception e){             e.printStackTrace();         }finally {             lock.unlock();         }     } }

生产者：Produce

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22

public class Produce implements Runnable{     private Handler handler;     public Produce(Handler handler) {         this.handler = handler;     }     @Override     public void run() {         for(int i = 0 ; i < 10 ; i++){             try {                 Thread.sleep(new Random().nextInt(1000));             } catch (InterruptedException e) {                 e.printStackTrace();             }             handler.put(String.valueOf(i));         }     } }

消费者：Consume

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

public class Consume implements Runnable{     private Handler handler;     public Consume(Handler handler) {         this.handler = handler;     }     @Override     public void run() {         for (int i = 0 ; i < 10 ; i++){             try {                 Thread.sleep(new Random().nextInt(1000));             } catch (InterruptedException e) {                 e.printStackTrace();             }             handler.eat();         }     } }

Main:

1 2 3 4 5 6 7 8 9 10 11 12 13 14

public class Main {     public static void main(String[] args) {         Handler handler = new Handler();         Produce produce = new Produce(handler);         Consume consume = new Consume(handler);         new Thread(consume).start();         new Thread(produce).start();         new Thread(produce).start();     } }

本文出自 “学海无涯 心境无限” 博客，请务必保留此出处http://zhangfengzhe.blog.51cto.com/8855103/1883655