java多线程教程之如何使用线程池详解

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime, 
    TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory，
    RejectedExecutionHandler handler){
     //...
}

exec.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

/**
 * 设置信号量的上界设置为线程池的大小加上可排队任务的数量，控制正在执行和等待执行的任务数量。
 */
public class BoundedExecutor {
 
 private final Executor exec;
 private final Semaphore semaphore;
 
 public BoundedExecutor(Executor exec,int bound){
  this.exec = exec;
  this.semaphore = new Semaphore(bound);
 }
 
 public void submitTask(final Runnable task) throws InterruptedException{
  semaphore.acquire();
  try{
   exec.execute(new Runnable(){
    public void run(){
     try{
      task.run();
     }finally{
      semaphore.release();
     }
    }
   });
  }catch(RejectedExecutionException e){
   semaphore.release();
  }
 }
}

public interface ThreadFactory{
 Thread newThread(Runnable r);
}

// 示例：将一个特定于线程池的名字传递给MyThread的构造函数，从而可以再线程转储和错误日志信息中区分来自不同线程池的线程。
 public class MyThreadFactory implements ThreadFactory{
 
 private final String poolName;
 
 public MyThreadFactory(String poolName){
  this.poolName = poolName;
 }
 
 public Thread newThread(Runnable runnable){
  return new MyThread(runnable,poolName);
 }
}

// 示例：为线程指定名字，设置自定义UncaughtExceptionHandler向Logger中写入信息及维护一些统计信息以及在线程被创建或者终止时把调试消息写入日志。
public class MyThread extends Thread{
 public static final String default_name = "myThread";
 private static volatile boolean debugLifecycle = false;
 private static final AtomicInteger created = new AtomicInteger();
 private static final AtomicInteger alive = new AtomicInteger();
 private static final Logger log = Logger.getAnonymousLogger();

 public MyThread(Runnable runnable){
  this(runnable,default_name);
 }

 public MyThread(Runnable runnable, String defaultName) {
  super(runnable,defaultName + "-" + created.incrementAndGet());
  setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
   @Override
   public void uncaughtException(Thread t, Throwable e) {
    log.log(Level.SEVERE,"uncaught in thread " + t.getName(), e); 

   }
  });
 }

 public void run(){
  boolean debug = debugLifecycle;
  if(debug){
   log.log(Level.FINE,"created " + getName());
  }
  try{
   alive.incrementAndGet();
   super.run();
  }finally{
   alive.decrementAndGet();
   if(debug){
    log.log(Level.FINE,"Exiting " + getName());
   }
  }

 }
}

/**
 * TimingThreadPool中给出了一个自定义的线程池，通过beforeExecute、afterExecute、terminated等方法来添加日志记录和统计信息收集。
 * 为了测量任务的运行时间，beforeExecute必须记录开始时间并把它保存到一个afterExecute可用访问的地方。
 * 因为这些方法将在执行任务的线程中调用，因此beforeExecute可以把值保存到一个ThreadLocal变量中。然后由afterExecute来取。
 * 在TimingThreadPool中使用了两个AtomicLong变量，分别用于记录已处理的任务和总的处理时间，并通过包含平均任务时间的日志消息。
 */
public class TimingThreadPool extends ThreadPoolExecutor{
 
 public TimingThreadPool(int corePoolSize, int maximumPoolSize,
   long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
  super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
 }
 
 private final ThreadLocal<Long> startTime = new ThreadLocal<Long>();
 private final Logger log = Logger.getLogger("TimingThreadPool");
 private final AtomicLong numTasks = new AtomicLong();
 private final AtomicLong totalTime = new AtomicLong();
 
 protected void beforeExecute(Thread t,Runnable r){
  super.beforeExecute(t, r);
  log.fine(String.format("Thread %s: start %s", t,r));
  startTime.set(System.nanoTime());
 }
 
 protected void afterExecute(Throwable t,Runnable r){
  try{
   long endTime = System.nanoTime();
   long taskTime = endTime - startTime.get();
   numTasks.incrementAndGet();
   totalTime.addAndGet(taskTime);
   log.fine(String.format("Thread %s: end %s, time=%dns", t,r,taskTime));
  }finally{
   super.afterExecute(r, t);
  }
 }
 
 protected void terminated(){
  try{
   log.info(String.format("Terminated: avg time=%dns", totalTime.get()/numTasks.get()));
  }finally{
   super.terminated();
  }
 }
}