详述：
线程阀是一种线程与线程之间相互制约和交互的机制；
作用：http://wsmajunfeng.iteye.com/blog/1629354
阻塞队列BlockingQueue；
数组阻塞队列ArrayBlockingQueue；
链表阻塞队列LinkedBlockingQueue；
优先级阻塞队列PriorityBlockingQueue；
延时队列DelayQueue；
同步队列SynchronousQueue；
链表双向阻塞队列LinkedBlockingDeque；
链表传输队列LinkedTransferQueue；
同步计数器CountDownLatch；
抽象队列化同步器AbstractQueuedSynchroizer；
同步计数器Semaphore；
同步计数器CyclicBarrier；
1 /** 2 * ArrayBlockingQueue的简单用法 3 */ 4 package thread04; 5 6 import java.util.concurrent.ArrayBlockingQueue; 7 import java.util.concurrent.BlockingQueue; 8 9 public class ArrayBlockingQueueTest01 10 { 11 public static void main(String[] args) throws InterruptedException 12 { 13 // 新建一个等待队列 14 final BlockingQueue<String> bq = new ArrayBlockingQueue<String>(16); 15 16 for(int i=0;i<16;i++) 17 { 18 String log = (i+1) + " --> "; 19 bq.put(log); 20 } 21 22 // 新建四个线程 23 for(int i=0;i<4;i++) 24 { 25 new Thread(new Runnable() 26 { 27 @Override 28 public void run() 29 { 30 while(true) 31 { 32 try 33 { 34 String log = bq.take(); 35 parseLog(log); 36 } 37 catch (InterruptedException e) 38 { 39 e.printStackTrace(); 40 } 41 } 42 } 43 }).start(); 44 } 45 } 46 47 public static void parseLog(String log) 48 { 49 System.out.println(log + System.currentTimeMillis()); 50 51 try 52 { 53 Thread.sleep(1000); 54 } 55 catch (InterruptedException e) 56 { 57 e.printStackTrace(); 58 } 59 } 60 }
1 /** 2 * LinkedBlockingQueue的简单用法 3 */ 4 package thread04; 5 6 import java.util.concurrent.BlockingQueue; 7 import java.util.concurrent.LinkedBlockingQueue; 8 9 public class LinkedBlockingQueueTest01 10 { 11 public static void main(String[] args) 12 { 13 final BlockingQueue<String> bq = new LinkedBlockingQueue<String>(16); 14 15 for(int i=0;i<16;i++) 16 { 17 String log = (i+1) + " --> "; 18 try 19 { 20 bq.put(log); 21 } 22 catch (InterruptedException e) 23 { 24 e.printStackTrace(); 25 } 26 } 27 28 for(int i=0;i<4;i++) 29 { 30 new Thread(new Runnable() 31 { 32 @Override 33 public void run() 34 { 35 while(true) 36 { 37 try 38 { 39 String log = bq.take(); 40 parseLog(log); 41 } 42 catch (InterruptedException e) 43 { 44 e.printStackTrace(); 45 } 46 } 47 } 48 49 }).start(); 50 } 51 } 52 53 public static void parseLog(String log) 54 { 55 System.out.println(log + System.currentTimeMillis()); 56 57 try 58 { 59 Thread.sleep(1000); 60 } 61 catch (InterruptedException e) 62 { 63 e.printStackTrace(); 64 } 65 } 66 }
1 /** 2 * DelayQueue的简单用法 3 */ 4 package thread04; 5 6 import java.util.concurrent.DelayQueue; 7 import java.util.concurrent.Delayed; 8 import java.util.concurrent.TimeUnit; 9 10 public class DelayQueueTest01 11 { 12 public static void main(String[] args) 13 { 14 // 新建一个等待队列 15 final DelayQueue<Student> dq = new DelayQueue<Student>(); 16 17 for(int i=0;i<5;i++) 18 { 19 Student student = new Student("学生"+i, Math.round((Math.random()*10+i))); 20 dq.put(student); // 将数据存到队列里面 21 } 22 23 // 获取但不移除此队列的头部；如果此队列为空，则返回null 24 System.out.println("dq.peek()：" + dq.peek().getName()); 25 26 // 获取并移除此队列的头部，在可从此队列获得到期延迟的元素，获得到达指定的等待时间之前一直等待（如有必要） 27 // poll(long timeout, TimeUnit unit) 28 } 29 } 30 31 class Student implements Delayed // 必须实现Delayed接口 32 { 33 private String name; 34 private long submitTime; // 交卷时间 35 private long workTime; // 考试时间 36 37 public String getName() 38 { 39 return this.name + "交卷，用时" + workTime; 40 } 41 42 public Student(String name, long submitTime) 43 { 44 this.name = name; 45 this.workTime = submitTime; 46 this.submitTime = TimeUnit.NANOSECONDS.convert(submitTime, TimeUnit.MILLISECONDS) + System.nanoTime(); 47 System.out.println(this.name + "交卷，用时" + workTime); 48 } 49 50 @Override 51 // 必须实现compareTo()方法 52 public int compareTo(Delayed o) 53 { 54 // 比较的方法 55 Student that = (Student) o; 56 return submitTime > that.submitTime ? 1 : (submitTime < that.submitTime ? -1 : 0); 57 } 58 59 @Override 60 // 必须实现getDelay()方法 61 public long getDelay(TimeUnit unit) 62 { 63 // 返回一个延时时间 64 return unit.convert(submitTime - System.nanoTime(), unit.NANOSECONDS); 65 } 66 67 } 68 69 /* 70 每次运行结果都不一样，我们获得永远是队列里面的第一个元素 71 */
1 /** 2 * SynchronousQueue的简单用法 3 */ 4 package thread04; 5 6 import java.util.concurrent.Semaphore; 7 import java.util.concurrent.SynchronousQueue; 8 9 public class SynchronousQueueTest01 10 { 11 public static void main(String[] args) 12 { 13 System.out.println("begin:" + System.currentTimeMillis() / 1000); 14 15 final SynchronousQueue<String> sq = new SynchronousQueue<String>(); 16 final Semaphore sem = new Semaphore(1); 17 18 /* 19 // 放在不同的地方，结果是不一样的 20 for(int i=0;i<10;i++) 21 { 22 String input = i + ""; 23 try 24 { 25 sq.put(input); 26 } 27 catch (InterruptedException e) 28 { 29 e.printStackTrace(); 30 } 31 } 32 */ 33 34 for(int i=0;i<10;i++) 35 { 36 new Thread(new Runnable() 37 { 38 @Override 39 public void run() 40 { 41 try 42 { 43 sem.acquire(); 44 45 String input = sq.take(); 46 String output = TestDo.doSome(input); 47 System.out.println(Thread.currentThread().getName() + ":" + output); 48 49 sem.release(); 50 } 51 catch (InterruptedException e) 52 { 53 e.printStackTrace(); 54 } 55 56 } 57 }).start(); 58 } 59 60 /*for(int i=0;i<10;i++) 61 { 62 String input = i + ""; 63 try 64 { 65 sq.put(input); 66 } 67 catch (InterruptedException e) 68 { 69 e.printStackTrace(); 70 } 71 }*/ 72 73 } 74 75 } 76 77 class TestDo 78 { 79 public static String doSome(String input) 80 { 81 try 82 { 83 Thread.sleep(1000); 84 } 85 catch (InterruptedException e) 86 { 87 e.printStackTrace(); 88 } 89 90 String output = input + ":" + System.currentTimeMillis(); 91 92 return output; 93 } 94 }
1 /** 2 * LinkedTransferQueue的简单使用 3 */ 4 package thread04; 5 6 import java.util.Random; 7 import java.util.concurrent.LinkedTransferQueue; 8 import java.util.concurrent.TimeUnit; 9 10 public class LinkedTransferQueueTest01 11 { 12 public static void main(String[] args) 13 { 14 LinkedTransferQueue<String> queue = new LinkedTransferQueue<String>(); 15 16 Producer p = new Producer(queue); 17 Thread producer = new Thread(p); 18 // 设为守护进程，使得线程执行结束后程序自动结束 19 producer.setDaemon(true); 20 producer.start(); 21 22 for(int i=0;i<10;i++) 23 { 24 Consumer c = new Consumer(queue); 25 Thread consumer = new Thread(c); 26 consumer.setDaemon(true); 27 consumer.start(); 28 29 try 30 { 31 // 消费者进程休眠1秒，以便生产者获得CPU，从而生产产品 32 Thread.sleep(1000); 33 } 34 catch (InterruptedException e) 35 { 36 e.printStackTrace(); 37 } 38 } 39 } 40 } 41 42 class Consumer implements Runnable 43 { 44 private LinkedTransferQueue<String> queue; 45 46 public Consumer(LinkedTransferQueue<String> queue) 47 { 48 this.queue = queue; 49 } 50 51 @Override 52 public void run() 53 { 54 try 55 { 56 System.out.println(" Consumer " + Thread.currentThread().getName() + " " + queue.take()); 57 } 58 catch (InterruptedException e) 59 { 60 e.printStackTrace(); 61 } 62 } 63 } 64 65 class Producer implements Runnable 66 { 67 private LinkedTransferQueue<String> queue; 68 69 public Producer(LinkedTransferQueue<String> queue) 70 { 71 this.queue = queue; 72 } 73 74 private String produce() 75 { 76 return " your lucky number " + (new Random().nextInt(1000)); 77 } 78 79 @Override 80 public void run() 81 { 82 try 83 { 84 while(true) 85 { 86 if(queue.hasWaitingConsumer()) 87 { 88 queue.transfer(produce()); 89 } 90 91 TimeUnit.SECONDS.sleep(1); 92 } 93 } 94 catch (Exception e) 95 { 96 e.printStackTrace(); 97 } 98 } 99 }
同步计数器CountDownLatch：
详解：
在完成一组正在其他线程中执行的操作之前，允许一个或多个线程一直等待；
用给定的计数初始化CountDownLatch；由于调用了countDown()方法，所以在当前计数到达0之前，awiait()方法会一直受阻塞（哪个线程中调用了await()方法，哪个线程就被阻塞，一直等待）；之后（当前计数到达0），会释放所有等待的线程，await的所有后续调用（线程调用await()方法后续的一些操作）都将立即返回；这种现象只出现一次，因为计数无法被重置；
主要方法：
CountDownLatch(int count) ：构造一个用给定计数初始化的CountDownLatch实例对象；
void await()：使当前线程在锁存器倒计数至0之前一直等待，除非线程被中断；
boolean await(long timeout, TimeUnit unit) ：使当前线程在锁存器倒计数至0之前一直等待，除非线程被中断或超出了指定的等待时间；
void countDown()：递减锁存器的计数，如果计数到达0，则释放所有等待的线程；
long getCount()：返回当前计数；
应用场景：
在一些应用场合中，需要等待某个条件到达要求后才能做后面的事情；同时当线程都完成后也会触发事件，以便进行后面的操作；
1 /** 2 * CountDownLatch的简单使用 3 */ 4 package thread04; 5 6 import java.util.concurrent.CountDownLatch; 7 8 public class CountDownLatchTest01 9 { 10 private static CountDownLatch latch = new CountDownLatch(3); 11 12 public static void main(String[] args) 13 { 14 Worker worker1 = new Worker("张三 程序员1", latch); 15 Worker worker2 = new Worker("李四 程序员2", latch); 16 Worker worker3 = new Worker("王五 程序员3", latch); 17 18 Thread t1 = new Thread(worker1); 19 Thread t2 = new Thread(worker2); 20 Thread t3 = new Thread(worker3); 21 22 t1.start(); 23 t2.start(); 24 t3.start(); 25 26 try 27 { 28 latch.await(); 29 } catch (InterruptedException e) 30 { 31 e.printStackTrace(); 32 } 33 34 System.out.println("Main is end!"); 35 } 36 } 37 38 class Worker implements Runnable 39 { 40 private String workerName; 41 private CountDownLatch latch; 42 43 public Worker(String workerName, CountDownLatch latch) 44 { 45 this.workerName = workerName; 46 this.latch = latch; 47 } 48 49 @Override 50 public void run() 51 { 52 try 53 { 54 System.out.println("Worker " + workerName + " is begin!"); 55 Thread.sleep(1000L); 56 System.out.println("Worker " + workerName + " is end!"); 57 58 } catch (InterruptedException e) 59 { 60 e.printStackTrace(); 61 } 62 latch.countDown(); 63 } 64 }
信号量Semaphore：
详解：
通过对信号量的不同操作，可以分别实现进程间的互斥与同步；
Semaphore可以控制某个资源被同时访问的任务数（维护了一个许可集合），它通过acquire()获取一个许可，release()释放一个许可；如果被同时访问的任务数已满，则其他acquire的任务进入等待状态，直到有一个任务被release掉，它才能得到许可；
使用场景：
排队场景，资源有限的房间，资源有限的群等等，常见的实际应用场景包括线程池、连接池等；
1 /** 2 * Semaphore的简单使用 3 */ 4 package thread04; 5 6 import java.util.concurrent.ExecutorService; 7 import java.util.concurrent.Executors; 8 import java.util.concurrent.Semaphore; 9 import java.util.concurrent.locks.ReentrantLock; 10 11 public class SemaphoreTest01 12 { 13 public static void main(String[] args) 14 { 15 ExecutorService es = Executors.newCachedThreadPool(); 16 17 final Semaphore sh = new Semaphore(5); 18 19 ReentrantLock rl = new ReentrantLock(); 20 21 for(int i=0;i<10;i++) 22 { 23 final int num = i; 24 25 Runnable run = new Runnable() 26 { 27 @Override 28 public void run() 29 { 30 // rl.lock(); 31 try 32 { 33 sh.acquire(); 34 System.out.println("线程 " + Thread.currentThread().getName() + " 获得许可：" + num); 35 for(int i=0;i<999999;i++); 36 sh.release(); 37 System.out.println("线程 " + Thread.currentThread().getName() + "释放许可：" + num); 38 System.out.println("当前允许进入的任务个数： " + sh.availablePermits()); 39 } 40 catch (InterruptedException e) 41 { 42 e.printStackTrace(); 43 } 44 finally 45 { 46 // rl.unlock(); 47 } 48 } 49 }; 50 es.execute(run); 51 } 52 53 es.shutdown(); 54 } 55 }
障碍器CyclicBarrier：
详解：
又叫同步计数器；
使用场景：
你希望创建一组任务，它们并发地执行工作，另外的一个任务在这一组任务并发执行结束前一直阻塞等待，直到该组任务全部执行结束，这个任务才得以执行；
1 /** 2 * CyclicBarrier的简单使用 3 */ 4 package thread04; 5 6 import java.util.concurrent.BrokenBarrierException; 7 import java.util.concurrent.CyclicBarrier; 8 9 public class CyclicBarrierTest01 10 { 11 public static void main(String[] args) 12 { 13 // 创建CyclicBarrier对象，并设置执行完一组5个线程的并发任务后，再执行MainTask任务 14 CyclicBarrier cb = new CyclicBarrier(5, new MainTask()); 15 16 SubTask sb1 = new SubTask("A", cb); 17 SubTask sb2 = new SubTask("B", cb); 18 SubTask sb3 = new SubTask("C", cb); 19 SubTask sb4 = new SubTask("D", cb); 20 SubTask sb5 = new SubTask("E", cb); 21 22 new Thread(sb1).start(); 23 new Thread(sb2).start(); 24 new Thread(sb3).start(); 25 new Thread(sb4).start(); 26 new Thread(sb5).start(); 27 28 } 29 } 30 31 /** 32 * 最后要执行的任务 33 * @author Administrator 34 * 35 */ 36 class MainTask implements Runnable 37 { 38 @Override 39 public void run() 40 { 41 System.out.println("前面的并发任务全部执行完毕后，开始执行最后任务..."); 42 } 43 } 44 45 /** 46 * 一组并发任务 47 * @author Administrator 48 * 49 */ 50 class SubTask implements Runnable 51 { 52 private String name; 53 private CyclicBarrier cb; 54 55 public SubTask(String name, CyclicBarrier cb) 56 { 57 this.name = name; 58 this.cb = cb; 59 } 60 61 @Override 62 public void run() 63 { 64 System.out.println("并发任务 " + name + " 开始执行..."); 65 for(int i=0;i<999999;i++); 66 System.out.println("并发任务 " + name + " 执行完毕，通知障碍器..."); 67 68 try 69 { 70 // 每执行完一项任务就通知障碍器 71 cb.await(); 72 } 73 catch (InterruptedException | BrokenBarrierException e) 74 { 75 e.printStackTrace(); 76 } 77 } 78 79 public String getName() 80 { 81 return name; 82 } 83 public void setName(String name) 84 { 85 this.name = name; 86 } 87 public CyclicBarrier getCb() 88 { 89 return cb; 90 } 91 public void setCb(CyclicBarrier cb) 92 { 93 this.cb = cb; 94 } 95 }
