JDK5.0 特性线程 同步装置之CountDownLatch 同步装置之CyclicBarrier 线程 BlockingQueue

 1 import java.util.concurrent.CountDownLatch;
 2 import java.util.concurrent.ExecutorService;
 3 import java.util.concurrent.Executors;
 4 
 5 /**
 6  * CountDownLatch维护一个计数器,等待这个CountDownLatch的线程必须等到计数器为0时才可以继续.
 7  * 以下实例模拟服务器的启动,假设启动一个服务器需要初始化3个组件,当3个组件初始化完毕后,服务器才算成功启动.
 8  */
 9 /**
10  * 使用CountDownLatch的关键技术点如下:
11  * 1.构造CountDownLatch对象时,需要指定计数器的初始值,该值必须大于等于0,一旦对象被创建,其初始值将不能被改变.
12  * 2.CountDownLatch的await方法使当前线程进入等待状态,直到计数器为0
13  * 3.CountDownLatch的 和countDown方法使计数器减1.
14  */
15 public class CountDownLatchTest {
16   /** 初始化组件的线程 */
17   public static class ComponentThread implements Runnable {
18     CountDownLatch latch; //计数器
19     int ID; //组件ID
20 
21     //构造方法
22     public ComponentThread(CountDownLatch latch, int ID) {
23       this.latch = latch;
24       this.ID = ID;
25     }
26 
27     public void run() {
28       //初始化组件
29       System.out.println("Initializing component " + ID);
30       try {
31         Thread.sleep(500 * ID);
32       }
33       catch (InterruptedException e) {}
34       System.out.println("Component " + ID + " initialized!");
35       latch.countDown(); //将计数器减1
36     }
37   }
38 
39   /** 启动服务器 */
40   public static void startServer() throws Exception {
41     System.out.println("Server is starting.");
42     //初始化一个初始值为3的CountDownLatch
43     CountDownLatch latch = new CountDownLatch(3);
44     //启动3个线程分别去3个组件
45     ExecutorService service = Executors.newCachedThreadPool();
46     service.submit(new ComponentThread(latch, 1));
47     service.submit(new ComponentThread(latch, 2));
48     service.submit(new ComponentThread(latch, 3));
49     service.shutdown();
50     latch.await();//等待3个组件的初始化工作都完成
51     System.out.println("Server is up!");//当所需的三个组件都完成时,Server就可以继续了
52   }
53 
54   public static void main(String... args) throws Exception {
55     CountDownLatchTest.startServer();
56   }
57 }
58 
59  

  1 import java.util.Random;
  2 
  3 import java.util.concurrent.CyclicBarrier;
  4 
  5  
  6 
  7 /**
  8 
  9  * CyclicBarrier维持一个计数器,与CountDownLatch不同的是,等待这个CyclicBarrier的线程必须等到计数器的某个值时,才可以继续.
 10 
 11  * CyclicBarrier就像它名字的意思一样,可看成是个障碍,所有的线程必须到齐后才能一起通过这个障碍.
 12 
 13  */
 14 
 15 /**
 16 
 17  * 本实例实现一个数组相邻元素的加法,一个线程给数组的第一个元素赋值,然后等待其它线程给数组第二个元素赋值,然后将第一个元素和第二个元素相加.
 18 
 19  */
 20 
 21 /**
 22 
 23  * CyclicBarrier的关键技术点如下:
 24 
 25  * 1.构造CyclicBarrier对象时,需要指定计数器的目标值,计数器的初始值为0.
 26 
 27  * 还可以在构造方法中带一个 Runnable参数,表示当计数器到达目标值是,在等待CyclicBarrier的线程被唤醒之前,指定该Runnable任务.
 28 
 29  * 2.CyclicBarrier的await方法使当前线程进入等待状态,同时将计数器值加1,当计数器到达目标值时,当前线程被唤醒.
 30 
 31  */
 32 
 33 public class CyclicBarrierTest {
 34 
 35        public static class ComponentThread implements Runnable{
 36 
 37               CyclicBarrier barrier;//计数器
 38 
 39               int ID;//组件
 40 
 41               int[] array; //数据数组
 42 
 43               public ComponentThread(CyclicBarrier barrier,int[] array,int ID){
 44 
 45                      this.barrier = barrier;
 46 
 47                      this.ID = ID;
 48 
 49                      this.array = array;
 50 
 51               }
 52 
 53               public void run(){
 54 
 55                      try{
 56 
 57                             //Random的nextInt(int n)方法返回一个[0,n)范围内的随机数
 58 
 59                             array[ID] = new Random().nextInt(100);
 60 
 61                             System.out.println("Componet " + ID + " sleep...");
 62 
 63                             barrier.await();
 64 
 65                             System.out.println("Componet " + ID + " awaked...");
 66 
 67                             //计算数据数组中的当前值和后续值
 68 
 69                             int result = array[ID] + array[ID + 1];
 70 
 71                             System.out.println("Component " + ID + " result: " + result);
 72 
 73                      }catch(Exception ex){
 74 
 75                      }
 76 
 77               }
 78 
 79        }
 80 
 81        /**测试CyclicBarrier的用法*/
 82 
 83        public static void testCyclicBarrier(){
 84 
 85               final int[] array = new int[3];
 86 
 87               CyclicBarrier barrier = new CyclicBarrier(2,new Runnable(){
 88 
 89                      public void run(){
 90 
 91                             System.out.println("testCyclicBarrier run...");
 92 
 93                             array[2] = array[0] + array[1];
 94 
 95                      }
 96 
 97               });
 98 
 99               //启动线程
100 
101               new Thread(new ComponentThread(barrier,array,0)).start();
102 
103               new Thread(new ComponentThread(barrier,array,1)).start();
104 
105        }
106 
107        public static void main(String... args){
108 
109               CyclicBarrierTest.testCyclicBarrier();
110 
111        }
112 
113 }

  1 importjava.util.concurrent.ArrayBlockingQueue;
  2 
  3 import java.util.concurrent.BlockingQueue;
  4 
  5 import java.util.concurrent.ExecutorService;
  6 
  7 import java.util.concurrent.Executors;
  8 
  9  
 10 
 11 /**
 12 
 13     本例介绍一个特殊的队列:BlockingQueue,如果BlockQueue是空的,从BlockingQueue取东西的操作将会被阻断进入等待状态,直到BlockingQueue进了东西才会被唤醒.同样,如果BlockingQueue是满的,任何试图往里存东西的操作也会被阻断进入等待状态,直到BlockingQueue里有空间才会被唤醒继续操作.
 14 
 15     本例再次实现11.4线程----条件Condition中介绍的篮子程序,不过这个篮子中最多能放的苹果数不是1,可以随意指定.当篮子满时,生产者进入等待状态,当篮子空时,消费者等待.
 16 
 17  */
 18 
 19 /**
 20 
 21     使用BlockingQueue的关键技术点如下:
 22 
 23     1.BlockingQueue定义的常用方法如下:
 24 
 25         1)add(anObject):把anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则招聘异常
 26 
 27         2)offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false.
 28 
 29         3)put(anObject):把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里面有空间再继续.
 30 
 31         4)poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null
 32 
 33         5)take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到Blocking有新的对象被加入为止
 34 
 35     2.BlockingQueue有四个具体的实现类,根据不同需求,选择不同的实现类
 36 
 37         1)ArrayBlockingQueue:规定大小的BlockingQueue,其构造函数必须带一个int参数来指明其大小.其所含的对象是以FIFO(先入先出)顺序排序的.
 38 
 39         2)LinkedBlockingQueue:大小不定的BlockingQueue,若其构造函数带一个规定大小的参数,生成的BlockingQueue有大小限制,若不带大小参数,所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定.其所含的对象是以FIFO(先入先出)顺序排序的
 40 
 41         3)PriorityBlockingQueue:类似于LinkedBlockQueue,但其所含对象的排序不是FIFO,而是依据对象的自然排序顺序或者是构造函数的Comparator决定的顺序.
 42 
 43         4)SynchronousQueue:特殊的BlockingQueue,对其的操作必须是放和取交替完成的.
 44 
 45     3.LinkedBlockingQueue和ArrayBlockingQueue比较起来,它们背后所用的数据结构不一样,导致LinkedBlockingQueue的数据吞吐量要大于ArrayBlockingQueue,但在线程数量很大时其性能的可预见性低于ArrayBlockingQueue.         
 46 
 47  */
 48 
 49 public class BlockingQueueTest {
 50 
 51        /**定义装苹果的篮子*/
 52 
 53        public static class Basket{
 54 
 55               //篮子,能够容纳3个苹果
 56 
 57               BlockingQueue<String> basket = new ArrayBlockingQueue<String>(3);
 58 
 59               //生产苹果,放入篮子
 60 
 61               public void produce() throws InterruptedException{
 62 
 63                      //put方法放入一个苹果,若basket满了,等到basket有位置
 64 
 65                      basket.put("An apple");
 66 
 67               }
 68 
 69               //消费苹果,从篮子中取走
 70 
 71               public String consume() throws InterruptedException{
 72 
 73                      //take方法取出一个苹果,若basket为空,等到basket有苹果为止
 74 
 75                      return basket.take();
 76 
 77               }
 78 
 79        }
 80 
 81        //测试方法
 82 
 83        public static void testBasket(){
 84 
 85               final Basket basket = new Basket();//建立一个装苹果的篮子
 86 
 87               //定义苹果生产者
 88 
 89               class Producer implements Runnable{
 90 
 91                      public void run(){
 92 
 93                             try{
 94 
 95                                    while(true){
 96 
 97                                           //生产苹果
 98 
 99                                           System.out.println("生产者准备生产苹果: " + System.currentTimeMillis());
100 
101                                           basket.produce();
102 
103                                           System.out.println("生产者生产苹果完毕: " + System.currentTimeMillis());
104 
105                                           //休眠300ms
106 
107                                           Thread.sleep(300);
108 
109                                    }
110 
111                             }catch(InterruptedException ex){
112 
113                             }
114 
115                      }
116 
117               }
118 
119               //定义苹果消费者
120 
121               class Consumer implements Runnable{
122 
123                      public void run(){
124 
125                             try{
126 
127                                    while(true){
128 
129                                           //消费苹果
130 
131                                           System.out.println("消费者准备消费苹果: " + System.currentTimeMillis());
132 
133                                           basket.consume();
134 
135                                           System.out.println("消费者消费苹果完毕: " + System.currentTimeMillis());
136 
137                                           //休眠1000ms
138 
139                                           Thread.sleep(1000);
140 
141                                    }
142 
143                             }catch(InterruptedException ex){
144 
145                             }
146 
147                      }
148 
149               }
150 
151               ExecutorService service = Executors.newCachedThreadPool();
152 
153               Producer producer = new Producer();
154 
155               Consumer consumer = new Consumer();
156 
157               service.submit(producer);
158 
159               service.submit(consumer);
160 
161               //程序运行5s后,所有任务停止
162 
163               try{
164 
165                      Thread.sleep(5000);
166 
167               }catch(InterruptedException ex){
168 
169               }
170 
171               service.shutdownNow();
172 
173        }
174 
175        public static void main(String[] args){
176 
177               BlockingQueueTest.testBasket();
178 
179        }
180 
181 }