Java 自定义线程池和线程总数控制操作

 public interface IThreadPool { /** * 关闭线程池 */ public void shutAlldown(); /** * 执行任务 * * @param job 任务 */ public void execute(Job job); /** * 添加工作者 * * @param addNum 添加数 */ public void addWorkers(int addNum); /** * 减少工作者 * * @param reduceNum 减少数目 */ public void reduceWorkers(int reduceNum); }

 import java.util.ArrayList; import java.util.Collections; import java.util.LinkedList; import java.util.List; public class XYThreadPool implements IThreadPool { // 默认线程数 private static int DEAFAULT_SIZE = 5; // 最大线程数 private static int MAX_SIZE = 10; // 任务列表 private LinkedList tasks = new LinkedList(); // 工作线程列表 private List workers = Collections .synchronizedList(new ArrayList()); /** * 默认构造函数 */ public XYThreadPool() { initWokers(DEAFAULT_SIZE); } /** * 执行线程数 * * @param threadNums 线程数 */ public XYThreadPool(int workerNum) { workerNum = workerNum <= 0 ? DEAFAULT_SIZE : workerNum > MAX_SIZE ? MAX_SIZE : workerNum; initWokers(workerNum); } /** * 初始化线程池 * * @param threadNums 线程数 */ public void initWokers(int threadNums) { for (int i = 0; i  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 { // 暂停指定数量的工作者 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) { // 任务队列为空 while (tasks.isEmpty()) { try { // 阻塞，放弃对象锁，等待被notify唤醒 tasks.wait(); System.out.println("block when tasks is empty"); } catch (InterruptedException e) { e.printStackTrace(); } } // 不为空取出任务 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 pool = new XYThreadPool(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 pool = new XYThreadPool(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; /** * 统一线程池管理类 */ @Component public class XYThreadManager { private XYThreadPool executorPool; @PostConstruct public void init() { executorPool = new XYThreadPool(10); } public XYThreadPool 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(); } }

 //使用无界队列 public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()); } //线程池数量是无限的 public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue()); }

 public class AsyncProcessQueue { // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ /** * Task 包装类
 * 此类型的意义是记录可能会被 Executor 吃掉的异常
 */ 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); } } } } // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ /** * 执行指定的任务 * * @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); /** * 默认最大并发数
 */ 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(); /** * 默认队列大小 */ private static final int DEFAULT_SIZE = 500; /** * 默认线程存活时间 */ private static final long DEFAULT_KEEP_ALIVE = 60L; /**NewEntryServiceImpl.java:689 * Executor */ private static ExecutorService executor; /** * 执行队列 */ private static BlockingQueue executeQueue = new ArrayBlockingQueue<>(DEFAULT_SIZE); static { // 创建 Executor // 此处默认最大值改为处理器数量的 4 倍 try { executor = new ThreadPoolExecutor(DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_CONCURRENT * 4, DEFAULT_KEEP_ALIVE, TimeUnit.SECONDS, executeQueue, FACTORY); // 关闭事件的挂钩 Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() { @Override public void run() { AsyncProcessor.LOGGER.info("AsyncProcessor shutting down."); executor.shutdown(); try { // 等待1秒执行关闭 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); } } /** * 此类型无法实例化 */ private AsyncProcessor() { } /** * 执行任务，不管是否成功
 * 其实也就是包装以后的 {@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; } /** * 提交任务，并可以在稍后获取其执行情况
 * 当提交失败时，会抛出 {@link } * * @param task * @return */ public static  Future submitTask(Callable 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 } });