public interface IThreadPool<Job extends Runnable> {
 /**
 * 關(guān)閉線程池
 */
 public void shutAlldown();
 
 /**
 * 執(zhí)行任務(wù)
 * 
 * @param job 任務(wù)
 */
 public void execute(Job job);
 
 /**
 * 添加工作者
 * 
 * @param addNum 添加數(shù)
 */
 public void addWorkers(int addNum);
 
 /**
 * 減少工作者
 * 
 * @param reduceNum 減少數(shù)目
 */
 public void reduceWorkers(int reduceNum);
}

import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List; 
public class XYThreadPool<Job extends Runnable> implements IThreadPool<Job> { 
 // 默認(rèn)線程數(shù)
 private static int DEAFAULT_SIZE = 5;
 // 最大線程數(shù)
 private static int MAX_SIZE = 10; 
 // 任務(wù)列表
 private LinkedList<Job> tasks = new LinkedList<Job>();
 // 工作線程列表
 private List<Worker> workers = Collections
  .synchronizedList(new ArrayList<Worker>()); 
 /**
 * 默認(rèn)構(gòu)造函數(shù)
 */
 public XYThreadPool() {
 initWokers(DEAFAULT_SIZE);
 } 
 /**
 * 執(zhí)行線程數(shù)
 * 
 * @param threadNums 線程數(shù)
 */
 public XYThreadPool(int workerNum) {
 workerNum = workerNum <= 0 ? DEAFAULT_SIZE
  : workerNum > MAX_SIZE ? MAX_SIZE : workerNum;
 initWokers(workerNum);
 } 
 /**
 * 初始化線程池
 * 
 * @param threadNums 線程數(shù)
 */
 public void initWokers(int threadNums) {
 for (int i = 0; i < threadNums; i++) {
  Worker worker = new Worker();
  worker.start();
  workers.add(worker);
 }
 // 添加關(guān)閉鉤子
 Runtime.getRuntime().addShutdownHook(new Thread() {
  public void run() {
  shutAlldown();
  }
 });
 } 
 @Override
 public void shutAlldown() {
 for (Worker worker : workers) {
  worker.shutdown();
 }
 } 
 @Override
 public void execute(Job job) {
 synchronized (tasks) {
  // 提交任務(wù)就是將任務(wù)對象加入任務(wù)隊(duì)列，等待工作線程去處理
  tasks.addLast(job);
  tasks.notifyAll();
 }
 } 
 @Override
 public void addWorkers(int addNum) {
 // 新線程數(shù)必須大于零，并且線程總數(shù)不能大于最大線程數(shù)
 if ((workers.size() + addNum) <= MAX_SIZE && addNum > 0) {
  initWokers(addNum);
 } else {
  System.out.println("addNum too large");
 }
 } 
 @Override
 public void reduceWorkers(int reduceNum) {
 if ((workers.size() - reduceNum <= 0))
  System.out.println("thread num too small");
 else {
  // 暫停指定數(shù)量的工作者
  int count = 0;
  while (count != reduceNum) {
  for (Worker w : workers) {
   w.shutdown();
   count++;
  }
  }
 }
 } 
 /**
 * 工作線程
 */
 class Worker extends Thread { 
 private volatile boolean flag = true; 
 @Override
 public void run() {
  while (flag) {
  Job job = null;
  // 加鎖（若只有一個woker可不必加鎖，那就是所謂的單線程的線程池，線程安全）
  synchronized (tasks) {
   // 任務(wù)隊(duì)列為空
   while (tasks.isEmpty()) {
   try {
    // 阻塞，放棄對象鎖，等待被notify喚醒
    tasks.wait();
    System.out.println("block when tasks is empty");
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
   }
   // 不為空取出任務(wù)
   job = tasks.removeFirst();
   System.out.println("get job:" + job + ",do biz");
   job.run();
  }
  }
 } 
 public void shutdown() {
  flag = false;
 }
 }
}

public class TestService1 {
 public static void main(String[] args) {
 // 啟動10個線程
 XYThreadPool<Runnable> pool = new XYThreadPool<Runnable>(10);
 pool.execute(new Runnable() {
  @Override
  public void run() {
  System.out.println("====1 test====");
  }
 }); 
 }
} 
public class TestService2 {
 public static void main(String[] args) {
 // 啟動10個線程
 XYThreadPool<Runnable> pool = new XYThreadPool<Runnable>(10);
 pool.execute(new Runnable() {
  @Override
  public void run() {
  System.out.println("====2 test====");
  }
 });
 }
}

import javax.annotation.PostConstruct;
import org.springframework.stereotype.Component;
import com.xy.pool.XYThreadPool; 
/**
 * 統(tǒng)一線程池管理類 
 */
@Component
public class XYThreadManager { 
 private XYThreadPool<Runnable> executorPool; 
 @PostConstruct
 public void init() {
 executorPool = new XYThreadPool<Runnable>(10);
 } 
 public XYThreadPool<Runnable> getExecutorPool() {
 return executorPool;
 }
} 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; 
@Service("testService3")
public class TestService3 { 
 @Autowired
 private XYThreadManager threadManager; 
 public void test() {
 threadManager.getExecutorPool().execute(new Runnable() {
  @Override
  public void run() {
  System.out.println("====3 test====");
  }
 });
 }
} 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; 
@Service("testService4")
public class TestService4 { 
 @Autowired
 private XYThreadManager threadManager; 
 public void test() {
 threadManager.getExecutorPool().execute(new Runnable() {
  @Override
  public void run() {
  System.out.println("====4 test====");
  }
 });
 }
} 
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext; 
public class TestMain { 
 @SuppressWarnings("resource")
 public static void main(String[] args) {
 ApplicationContext atc = new ClassPathXmlApplicationContext("applicationContext.xml"); 
 TestService3 t3 = (TestService3) atc.getBean("testService3");
 t3.test(); 
 TestService4 t4 = (TestService4) atc.getBean("testService4");
 t4.test();
 } 
}

//使用無界隊(duì)列
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                   0L, TimeUnit.MILLISECONDS,
                   new LinkedBlockingQueue<Runnable>());
  }
 
//線程池?cái)?shù)量是無限的
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                   60L, TimeUnit.SECONDS,
                   new SynchronousQueue<Runnable>());
  }

public class AsyncProcessQueue { 
 // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ 
 /**
 * Task 包裝類<br>
 * 此類型的意義是記錄可能會被 Executor 吃掉的異常<br>
 */
 public static class TaskWrapper implements Runnable {
 private static final Logger _LOGGER = LoggerFactory.getLogger(TaskWrapper.class); 
 private final Runnable gift; 
 public TaskWrapper(final Runnable target) {
  this.gift = target;
 } 
 @Override
 public void run() {
 
  // 捕獲異常，避免在 Executor 里面被吞掉了
  if (gift != null) {
 
  try {
   gift.run();
  } catch (Exception e) {
   _LOGGER.error("Wrapped target execute exception.", e);
  }
  }
 }
 } 
 // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 /**
 * 執(zhí)行指定的任務(wù)
 * 
 * @param task
 * @return
 */
 public static boolean execute(final Runnable task) {
 return AsyncProcessor.executeTask(new TaskWrapper(task));
 }
}
public class AsyncProcessor {
 static final Logger LOGGER = LoggerFactory.getLogger(AsyncProcessor.class);
 
 /**
 * 默認(rèn)最大并發(fā)數(shù)<br>
 */
 private static final int DEFAULT_MAX_CONCURRENT = Runtime.getRuntime().availableProcessors() * 2;
 
 /**
 * 線程池名稱格式
 */
 private static final String THREAD_POOL_NAME = "ExternalConvertProcessPool-%d";
 
 /**
 * 線程工廠名稱
 */
 private static final ThreadFactory FACTORY = new BasicThreadFactory.Builder().namingPattern(THREAD_POOL_NAME)
  .daemon(true).build();
 
 /**
 * 默認(rèn)隊(duì)列大小
 */
 private static final int DEFAULT_SIZE = 500;
 
 /**
 * 默認(rèn)線程存活時間
 */
 private static final long DEFAULT_KEEP_ALIVE = 60L;
 
 /**NewEntryServiceImpl.java:689
 * Executor
 */
 private static ExecutorService executor;
 
 /**
 * 執(zhí)行隊(duì)列
 */
 private static BlockingQueue<Runnable> executeQueue = new ArrayBlockingQueue<>(DEFAULT_SIZE); 
 static {
 // 創(chuàng)建 Executor
 // 此處默認(rèn)最大值改為處理器數(shù)量的 4 倍
 try {
  executor = new ThreadPoolExecutor(DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_CONCURRENT * 4, DEFAULT_KEEP_ALIVE,
   TimeUnit.SECONDS, executeQueue, FACTORY); 
  // 關(guān)閉事件的掛鉤
  Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() { 
  @Override
  public void run() {
   AsyncProcessor.LOGGER.info("AsyncProcessor shutting down."); 
   executor.shutdown(); 
   try { 
   // 等待1秒執(zhí)行關(guān)閉
   if (!executor.awaitTermination(1, TimeUnit.SECONDS)) {
    AsyncProcessor.LOGGER.error("AsyncProcessor shutdown immediately due to wait timeout.");
    executor.shutdownNow();
   }
   } catch (InterruptedException e) {
   AsyncProcessor.LOGGER.error("AsyncProcessor shutdown interrupted.");
   executor.shutdownNow();
   } 
   AsyncProcessor.LOGGER.info("AsyncProcessor shutdown complete.");
  }
  }));
 } catch (Exception e) {
  LOGGER.error("AsyncProcessor init error.", e);
  throw new ExceptionInInitializerError(e);
 }
 } 
 /**
 * 此類型無法實(shí)例化
 */
 private AsyncProcessor() {
 } 
 /**
 * 執(zhí)行任務(wù)，不管是否成功<br>
 * 其實(shí)也就是包裝以后的 {@link Executer} 方法
 * 
 * @param task
 * @return
 */
 public static boolean executeTask(Runnable task) { 
 try {
  executor.execute(task);
 } catch (RejectedExecutionException e) {
  LOGGER.error("Task executing was rejected.", e);
  return false;
 } 
 return true;
 } 
 /**
 * 提交任務(wù)，并可以在稍后獲取其執(zhí)行情況<br>
 * 當(dāng)提交失敗時，會拋出 {@link }
 * 
 * @param task
 * @return
 */
 public static <T> Future<T> submitTask(Callable<T> task) { 
 try {
  return executor.submit(task);
 } catch (RejectedExecutionException e) {
  LOGGER.error("Task executing was rejected.", e);
  throw new UnsupportedOperationException("Unable to submit the task, rejected.", e);
 }
 }
}

AsyncProcessQueue.execute(new Runnable() {
     @Override
     public void run() {
        //do something
    }
});