首页 > 代码库 > Java的并发神器concurrent包详解（一）
Java的并发神器concurrent包详解（一）
在JDK 1.5之前，提到并发，java程序员们一般想到的是wait()、notify()、Synchronized关键字等，但是并发除了要考虑竞态资源、死锁、资源公平性等问题，往往还需要考虑性能问题，在一些业务场景往往还会比较复杂，这些都给java coder们造成不小的难题。JDK 1.5的concurrent包帮我们解决了不少问题。
Concurrent包中包含了几个比较常用的并发模块，这个系列，LZ就和大家一起来学习各个模块，Let’s Go!
一、线程池的基本用法
一般并发包里有三个常用的线程池实例化方法，在Executors这个工厂类中。
- ·newFixedThreadPool(int size):
创建一个可重用固定线程数的线程池，以共享的无界队列（LinkedBlockingQueue）方式来运行这些线程。任何时刻最多只有size大小的线程在执行各自任务。当线程池中所有线程都在运行，后续新来的线程将在无界队列中等待，直到有空闲线程为止。如果有线程在执行任务期间终止退出，一个新的线程会替代原先的线程继续后面任务的执行。线程池中的线程会一直存在，除非显示的关闭某个线程。
ExecuterService fixedPool = Executors. newFixedThreadPool(5); //创建一个固定包含5个线程的线程池
- ·newCachedThreadPool()
创建一个可以根据需要创建线程的线程池，当一个任务过来，如果线程池中有空闲可用线程，先用空闲线程，没有就创建一个新线程。这种线程池对于执行short-lived asynchronous tasks（短期异步任务）通常会提高性能。默认空闲60s的线程会从线程池中移除，如果长时间空闲的话，这个线程池是不会占用任何资源的。但是以我个人的经验，这种线程池要慎用，用不好经常会造成堆外内存溢出，从而造成应用反应慢，最后导致宕机。
ExecuterService fixedPool = Executors. newCachedThreadPool();
- ·newSingleThreadExecutor()
创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程。可以保证顺序执行各个任务。有人说这不就是newFixedThreadPool(1)么，错矣。看源码就知道外部还包了一层FinalizableDelegatedExecutorService，这个是Executors的一个静态内部类，这个内部类继承了DelegatedExecutorService，返回的这个ExecutorService实现只包含ExecutorService接口定义的一些方法，不同于newFixedThreadPool有自己扩展的方法。源码注释的区别是：newSingleThreadExecutor可以保证使用其他线程时，返回的Executor不用重新配置。
ExecuterService fixedPool = Executors. newSingleThreadExecutor();
- ·newScheduledThreadPool(int size)
创建一个线程池，它可安排在给定延迟后运行命令或者定期地执行。这是个无界大小的线程池。其中主要的执行方法是scheduleAtFixedRate和scheduleWithFixedDelay方法，最终的核心方法是delayedExecute，这两个方法有什么区别呢，简单来说，scheduleAtFixedRate是以任务执行时间开始为延迟时间起点的，scheduleWithFixedDelay是以任务执行结束时间为延时时间起点的。
接下来对并发包里面常用的一些类进行使用解析
1） 信号量 Semaphore:
一个计数信号量，维护了一个许可集。通俗的讲，就是对有限共享资源记录使用情况。
举个日常生活中的例子好了：有20个人去登记信息表格，但是只有3只笔，一个人用完之后给下面一个人用，直到所有人都用完。
没有并发包之前，我们是怎么设计的呢?
定义一个变量a，用synchronized修饰，值为3，新建20个线程模拟20个人访问，每个人用笔之前，先看是否有空余的笔，如果有，就使用，并将变量a减少1，使用完成之后，再放回去，即将变量a加1，各个线程之间对于变量a的访问需要互斥进行。Synchronized是一个重量级的同步操作，涉及到用户态与核心态的切换，所以性能一般。
让我们再看看信号量是怎么实现的，Semaphore实现的思想跟上面差不多。只不过Semaphore已经帮我们做了同步处理，有两个关键的方法：
- ·acquire() 获取一个信号量许可
- ·release() 释放一个信号量许可
此外还有其他一些方法：availablePermits() 返回信号量可用的许可数，talk is cheap show me the code：
public class SemaphoreTest extends Thread{
private String name;
private Semaphore sh;
public SemaphoreTest(String name,Semaphore sh){
this.name = name;
this.sh = sh;
}
public void run(){
if(sh.availablePermits()>0){
System.out.println("有笔");
}else{
System.out.println("笔没了，等等");
}
try {
sh.acquire();//信号量减1
System.out.println(this.name+"号在用笔");
Thread.sleep((long) (Math.random()*1000));
sh.release();
System.out.println(this.name+"号用完了");
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public static void main(String[] args) {
ExecutorService es = Executors.newCachedThreadPool();
Semaphore sh = new Semaphore(3);
for(int i=0;i<20;i++){
es.submit(new SemaphoreTest(i+"",sh));
}
es.shutdown();
}
}
上述代码直接运行即可。
这个信号量是如何同步资源的，就需要读读jdk相关源码了。核心类是AbstractQueuedSynchronizer，该类实现了一个FIFO的列表，列表中Node表示列表节点，该Node有prev、next、Thread等属性，表示前节点后节点以及线程等。并发包中每个需要实现个性化同步机制的都要扩展该类，以实现不同的同步功能。比如Semaphore、countDownLatch、CyclicBarrier等，具体原理要分析源码，篇幅较长，这里不做展开，有兴趣的可以研究一下，对volatile关键字、CAS等都会有更深的理解。
2） CountDownLatch
用给定的计数 初始化 CountDownLatch。由于调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。
比较重要的2个方法：countDown()和await()
下面来个常见的例子：导游带了20人团到了饭店吃饭，要等到20个人到齐了，才开始吃饭，还要等到20人都吃完了，才可以继续下个景点。
来看看不用并发包你会怎么实现，首先会设置20个人到来以及吃完的标志，每个线程过来更改自己的标志，主程序for或者while不停的循环监听，制止全部到来以及吃完才继续。如果是100人1000人呢，未免臃肿。
CountDownLatch的实现：
public class CountDownLautchTest {
public static void main(String[] args) {
ExecutorService es = Executors.newFixedThreadPool(20);
final CountDownLatch endFlag = new CountDownLatch(20);
final CountDownLatch startFlag = new CountDownLatch(1);
final CountDownLatch comeFlag = new CountDownLatch(20);
for(int i=0;i<20;i++){
final int j = i + 1;
Runnable person = new Runnable(){
@Override
public void run() {
System.out.println(j+"号游客来了");
comeFlag.countDown();
try {
startFlag.await();
Thread.sleep(1000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
System.out.println(j+"号吃完了");
endFlag.countDown();
}
}
};
es.submit(person);
}
try {
comeFlag.await(); //保证所有人都到齐了
System.out.println("人都齐了，大家一起吃饭");
startFlag.countDown();//开吃
endFlag.await();//等待所有人都吃完了
System.out.println("全都吃完了，继续下个景点");
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
es.shutdown();
}
}
CountDownLatch是一种不可逆的降序计数操作，所以上述代码里面定义了2个分别表示等候以及吃饭的标志。如果需要重复计数，就需要用到下面这个类：CyclicBarrier。
3） CyclicBarrier
一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。
还是上述的例子吧，再加一点场景，吃完饭还要游玩，游玩结束返程回家。让我们来看看CyclicBarrier是如何实现的，上代码：
public class CyclicBarrierTest {
public static void main(String[] args) {
final CyclicBarrier barrier = new CyclicBarrier(20);
final ExecutorService es = Executors.newFixedThreadPool(20);
System.out.println("****等人齐****");
for(int i=0;i<20;i++){
final int j = i+1;
Runnable person = new Runnable(){
@Override
public void run() {
try {
System.out.println(j+"号来了");
if(barrier.await()==0){
System.out.println("****人都到齐了****");
}
System.out.println(j+"号开吃");
if(barrier.await()==0){
System.out.println("****吃完了出发****");
}
System.out.println(j+"号玩好了");
if(barrier.await()==0){
System.out.println("****游玩结束回家****");
}
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
};
es.submit(person);
}
es.shutdown();
}
}
代码直接运行即可，应该比较好懂吧，这里有个小窍门，代码中if(barrier.await()==0)这个条件，有点类似于CountDownLatch的await往下执行的条件了，每次判断都会减1，减到0，即完成所有线程的等待，继续下面的操作。
结语
关于并发包一些常用的类就介绍到这里吧，工作中具体怎么使用要具体分析各个业务场景，选择合适的方法，随机应变，触类旁通吧，后续会介绍一下Lock源码以及实现，涉及AQS的原理等。
未完，待续……
Java的并发神器concurrent包详解（一）