java concurrent 探秘(1)

        我们都知道，在JDK1.5之前，Java中要进行业务并发时，通常需要有程序员独立完成代码实现，当然也有一些开源的框架提供了这些功能，但是这些依然没有JDK自带的功能使用起来方便。而当针对高质量Java多线程并发程序设计时,为防止死蹦等现象的出现，比如使用java之前的wait()、notify()和synchronized等，每每需要考虑性能、死锁、公平性、资源管理以及如何避免线程安全性方面带来的危害等诸多因素，往往会采用一些较为复杂的安全策略，加重了程序员的开发负担.万幸的是，在JDK1.5出现之后，Sun大神（Doug Lea）终于为我们这些可怜的小程序员推出了java.util.concurrent工具包以简化并发完成。开发者们借助于此，将有效的减少竞争条件（race conditions）和死锁线程。concurrent包很好的解决了这些问题，为我们提供了更实用的并发程序模型。


Executor                 ：具体Runnable任务的执行者。
ExecutorService          ：一个线程池管理者，其实现类有多种，我会介绍一部分。我们能把Runnable,Callable提交到池中让其调度。
Semaphore                ：一个计数信号量
ReentrantLock            ：一个可重入的互斥锁定 Lock，功能类似synchronized，但要强大的多。
Future                   ：是与Runnable,Callable进行交互的接口，比如一个线程执行结束后取返回的结果等等，还提供了cancel终止线程。
BlockingQueue            ：阻塞队列。
CompletionService        : ExecutorService的扩展，可以获得线程执行结果的
CountDownLatch           ：一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。
CyclicBarrier            ：一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点
Future                   ：Future 表示异步计算的结果。
ScheduledExecutorService ：一个 ExecutorService，可安排在给定的延迟后运行或定期执行的命令。

接下来逐一介绍

Executors主要方法说明
newFixedThreadPool（固定大小线程池）
创建一个可重用固定线程集合的线程池，以共享的无界队列方式来运行这些线程（只有要请求的过来，就会在一个队列里等待执行）。如果在关闭前的执行期间由于失败而导致任何线程终止，那么一个新线程将代替它执行后续的任务（如果需要）。

newCachedThreadPool（无界线程池，可以进行自动线程回收）
创建一个可根据需要创建新线程的线程池，但是在以前构造的线程可用时将重用它们。对于执行很多短期异步任务的程序而言，这些线程池通常可提高程序性能。调用 execute 将重用以前构造的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。因此，长时间保持空闲的线程池不会使用任何资源。注意，可以使用 ThreadPoolExecutor 构造方法创建具有类似属性但细节不同（例如超时参数）的线程池。

newSingleThreadExecutor（单个后台线程）
创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程。（注意，如果因为在关闭前的执行期间出现失败而终止了此单个线程，那么如果需要，一个新线程将代替它执行后续的任务）。可保证顺序地执行各个任务，并且在任意给定的时间不会有多个线程是活动的。与其他等效的 newFixedThreadPool(1) 不同，可保证无需重新配置此方法所返回的执行程序即可使用其他的线程。

这些方法返回的都是ExecutorService对象，这个对象可以理解为就是一个线程池。
这个线程池的功能还是比较完善的。可以提交任务submit()可以结束线程池shutdown()。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MyExecutor extends Thread {
private int index;
public MyExecutor(int i){
   this.index=i;
}
public void run(){
   try{
    System.out.println("["+this.index+"] start....");
    Thread.sleep((int)(Math.random()*1000));
    System.out.println("["+this.index+"] end.");
   }
   catch(Exception e){
    e.printStackTrace();
   }
}

public static void main(String args[]){
   ExecutorService service=Executors.newFixedThreadPool(4);
   for(int i=0;i<10;i++){
    service.execute(new MyExecutor(i));
    //service.submit(new MyExecutor(i));
   }
   System.out.println("submit finish");
   service.shutdown();
}
}

虽然打印了一些信息，但是看的不是非常清晰，这个线程池是如何工作的，我们来将休眠的时间调长10倍。
Thread.sleep((int)(Math.random()*10000));

再来看，会清楚看到只能执行4个线程。当执行完一个线程后，才会又执行一个新的线程，也就是说，我们将所有的线程提交后，线程池会等待执行完最后shutdown。我们也会发现，提交的线程被放到一个“无界队列里”。这是一个有序队列（BlockingQueue，这个下面会说到）。

另外它使用了Executors的静态函数生成一个固定的线程池，顾名思义，线程池的线程是不会释放的，即使它是Idle。
这就会产生性能问题，比如如果线程池的大小为200，当全部使用完毕后，所有的线程会继续留在池中，相应的内存和线程切换（while(true)+sleep循环）都会增加。
如果要避免这个问题，就必须直接使用ThreadPoolExecutor()来构造。可以像通用的线程池一样设置“最大线程数”、“最小线程数”和“空闲线程keepAlive的时间”。


这个就是线程池基本用法。

Semaphore
一个计数信号量。从概念上讲，信号量维护了一个许可集合。如有必要，在许可可用前会阻塞每一个 acquire()，然后再获取该许可。每个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。

Semaphore 通常用于限制可以访问某些资源（物理或逻辑的）的线程数目。例如，下面的类使用信号量控制对内容池的访问：

这里是一个实际的情况，大家排队上厕所，厕所只有两个位置，来了10个人需要排队。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class MySemaphore extends Thread {
Semaphore position;
private int id;
public MySemaphore(int i,Semaphore s){
   this.id=i;
   this.position=s;
}

public void run(){
   try{
    if(position.availablePermits()>0){
     System.out.println("顾客["+this.id+"]进入厕所，有空位");
    }
    else{
     System.out.println("顾客["+this.id+"]进入厕所，没空位，排队");
    }
    position.acquire();
    System.out.println("顾客["+this.id+"]获得坑位");
    Thread.sleep((int)(Math.random()*1000));
    System.out.println("顾客["+this.id+"]使用完毕");
    position.release();
   }
   catch(Exception e){
    e.printStackTrace();
   }
}
public static void main(String args[]){
   ExecutorService list=Executors.newCachedThreadPool();
   Semaphore position=new Semaphore(2);
   for(int i=0;i<10;i++){
    list.submit(new MySemaphore(i+1,position));
   }
   list.shutdown();
   position.acquireUninterruptibly(2);
   System.out.println("使用完毕，需要清扫了");
   position.release(2);
}
}

ReentrantLock
一个可重入的互斥锁定 Lock，它具有与使用 synchronized 方法和语句所访问的隐式监视器锁定相同的一些基本行为和语义，但功能更强大。

ReentrantLock 将由最近成功获得锁定，并且还没有释放该锁定的线程所拥有。当锁定没有被另一个线程所拥有时，调用 lock 的线程将成功获取该锁定并返回。如果当前线程已经拥有该锁定，此方法将立即返回。可以使用 isHeldByCurrentThread() 和 getHoldCount() 方法来检查此情况是否发生。

此类的构造方法接受一个可选的公平参数。
当设置为 true时，在多个线程的争用下，这些锁定倾向于将访问权授予等待时间最长的线程。否则此锁定将无法保证任何特定访问顺序。
与采用默认设置（使用不公平锁定）相比，使用公平锁定的程序在许多线程访问时表现为很低的总体吞吐量（即速度很慢，常常极其慢），但是在获得锁定和保证锁定分配的均衡性时差异较小。不过要注意的是，公平锁定不能保证线程调度的公平性。因此，使用公平锁定的众多线程中的一员可能获得多倍的成功机会，这种情况发生在其他活动线程没有被处理并且目前并未持有锁定时。还要注意的是，未定时的 tryLock 方法并没有使用公平设置。因为即使其他线程正在等待，只要该锁定是可用的，此方法就可以获得成功。

建议总是 立即实践，使用 try 块来调用 lock，在之前/之后的构造中，最典型的代码如下：

class X {
   private final ReentrantLock lock = new ReentrantLock();
   // ...

   public void m() { 
     lock.lock(); // block until condition holds
     try {
       // ... method body
     } finally {
       lock.unlock()
     }
   }
}

我的例子：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReentrantLock;

public class MyReentrantLock extends Thread{
TestReentrantLock lock;
private int id;
public MyReentrantLock(int i,TestReentrantLock test){
   this.id=i;
   this.lock=test;
}

public void run(){
   lock.print(id);
}

public static void main(String args[]){
   ExecutorService service=Executors.newCachedThreadPool();
   TestReentrantLock lock=new TestReentrantLock();
   for(int i=0;i<10;i++){
    service.submit(new MyReentrantLock(i,lock));
   }
   service.shutdown();
}
}
class TestReentrantLock{
private ReentrantLock lock=new ReentrantLock();
public void print(int str){
   try{
    lock.lock();
    System.out.println(str+"获得");
    Thread.sleep((int)(Math.random()*1000));
   }
   catch(Exception e){
    e.printStackTrace();
   }
   finally{
    System.out.println(str+"释放");
    lock.unlock();
   }
}
}