java多线程-线程池使用-元一软件 前言线程池是Java中使用较多的并发框架合理使用线程池可以降低资源消耗提高响应速度提高线程的可管理性。本篇文章为《Java并发编程的艺术》第9章的学习笔记根据原文作者的编写思路依次对线程池的原理线程池的创建线程池执行任务和关闭线程池进行了学习和总结。最后会给出一个线程池的实现案例该案例为之前在某微信公众号上看到的线程池实现实战已经无法考究其出处原案例对于线程池的关闭存在缺陷我对其进行了一些修正和说明分享出来一起学习。正文一. 线程池的原理当一个任务提交到线程池ThreadPoolExecutor时该任务的执行如下图所示。如果当前运行的线程数小于corePoolSzie核心线程数则创建新线程来执行任务需要获取全局锁如果当前运行的线程数等于或大于corePoolSzie则将任务加入BlockingQueue任务阻塞队列如果BlockingQueue已满则创建新的线程来执行任务需要获取全局锁如果创建新线程会使当前线程数大于maximumPoolSize最大线程数则拒绝任务并调用RejectedExecutionHandler的rejectedExecution()方法。由于ThreadPoolExecutor存储工作线程使用的集合是HashSet因此执行上述步骤1和步骤3时需要获取全局锁来保证线程安全而获取全局锁会导致线程池性能瓶颈因此通常情况下线程池完成预热后当前线程数大于等于corePoolSize线程池的execute()方法都是执行步骤2。二. 线程池的创建通过ThreadPoolExecutor能够创建一个线程池ThreadPoolExecutor的构造函数签名如下。public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue) public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, ThreadFactory threadFactory) public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, RejectedExecutionHandler handler) public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)通过ThreadPoolExecutor创建线程池时需要指定线程池的核心线程数最大线程数线程保活时间线程保活时间单位和任务阻塞队列并按需指定线程工厂和饱和拒绝策略如果不指定线程工厂和饱和拒绝策略则ThreadPoolExecutor会使用默认的线程工厂和饱和拒绝策略。下面分别介绍这些参数的含义。参数含义corePoolSize核心线程数即线程池的基本大小。当一个任务被提交到线程池时如果线程池的线程数小于corePoolSize那么无论其余线程是否空闲也需创建一个新线程来执行任务。maximumPoolSize最大线程数。当线程池中线程数大于等于corePoolSize时新提交的任务会加入任务阻塞队列但是如果任务阻塞队列已满且线程数小于maximumPoolSize此时会继续创建新的线程来执行任务。该参数规定了线程池允许创建的最大线程数keepAliveTime线程保活时间。当线程池的线程数大于核心线程数时多余的空闲线程会最大存活keepAliveTime的时间如果超过这个时间且空闲线程还没有获取到任务来执行则该空闲线程会被回收掉。unit线程保活时间单位。通过TimeUnit指定线程保活时间的时间单位可选单位有DAYS天HOURS时MINUTES分SECONDS秒MILLISECONDS毫秒MICROSECONDS微秒和NANOSECONDS纳秒但无论指定什么时间单位ThreadPoolExecutor统一会将其转换为NANOSECONDS。workQueue任务阻塞队列。线程池的线程数大于等于corePoolSize时新提交的任务会添加到workQueue中所有线程执行完上一个任务后会循环从workQueue中获取任务来执行。threadFactory创建线程的工厂。可以通过线程工厂给每个创建出来的线程设置更有意义的名字。handler饱和拒绝策略。如果任务阻塞队列已满且线程池中的线程数等于maximumPoolSize说明线程池此时处于饱和状态应该执行一种拒绝策略来处理新提交的任务。三. 线程池执行任务线程池使用两个方法执行任务分别为execute()和submit()。execute()方法用于执行不需要返回值的任务submit()方法用于执行需要返回值的任务。execute()是接口Executor定义的方法submit()是接口ExecutorService定义的方法相关类图如下所示。ThreadPoolExecutor对execute()的实现如下。public void execute(Runnable command) { if (command null) throw new NullPointerException(); int c ctl.get(); if (workerCountOf(c) corePoolSize) { if (addWorker(command, true)) return; c ctl.get(); } if (isRunning(c) workQueue.offer(command)) { int recheck ctl.get(); if (! isRunning(recheck) remove(command)) reject(command); else if (workerCountOf(recheck) 0) addWorker(null, false); } else if (!addWorker(command, false)) reject(command); }AbstractExecutorService对submit()实现如下。public Future? submit(Runnable task) { if (task null) throw new NullPointerException(); RunnableFutureVoid ftask newTaskFor(task, null); execute(ftask); return ftask; } public T FutureT submit(Runnable task, T result) { if (task null) throw new NullPointerException(); RunnableFutureT ftask newTaskFor(task, result); execute(ftask); return ftask; } public T FutureT submit(CallableT task) { if (task null) throw new NullPointerException(); RunnableFutureT ftask newTaskFor(task); execute(ftask); return ftask; }在execute()方法中会根据当前线程数决定是新建线程来处理任务还是添加任务到任务阻塞队列中而在submit()方法中是将任务封装成RunnableFuture然后再调用execute()方法。四. 关闭线程池可以通过调用线程池的shutdown()或者shutdownNow()方法来关闭线程池。shutdown()方法会将线程池状态置为SHUTDOWN此时线程池不会再接收新提交的任务空闲的线程会被中断当正在被执行的任务和任务阻塞队列中的任务执行完后线程池才会安全的关闭掉。shutdownNow()方法会将线程池状态置为STOP此时线程池不会再接收新提交的任务所有线程会被中断任务阻塞队列中的任务不再执行这些任务会以列表形式返回正在执行中的任务也会被尝试停止。补充上述中的空闲线程可以理解为正在从任务阻塞队列中获取任务的线程即没有在执行任务的线程。五. 线程池实现实战在ThreadPoolExecutor中存储Worker工作线程的集合为HashSet因此每次对Worker集合做操作时需要获取全局锁。在本线程池实现的实战中将基于ConcurrentHashMap实现一个ConcurrentHashSet并作为存储Worker的集合。实现如下。public class ConcurrentHashSetT extends AbstractSetT { private final ConcurrentHashMapT, Object MAP new ConcurrentHashMap(); // 无实际意义用于和Worker组成键值对存入ConcurrentHashMap private final Object PRESENT new Object(); // 用于统计当前集合中的Worker数量 private final AtomicInteger COUNT new AtomicInteger(); Override public boolean add(T t) { COUNT.incrementAndGet(); return MAP.put(t, PRESENT) null; } Override public boolean remove(Object o) { COUNT.decrementAndGet(); return MAP.remove(o) PRESENT; } Override public IteratorT iterator() { return MAP.keySet().iterator(); } Override public int size() { return COUNT.get(); } }再看一下ThreadPool的字段。public class ThreadPool { // 全局锁 private final ReentrantLock lock new ReentrantLock(); // 核心线程数量 private final int coreSize; // 最大线程数量 private final int maxSize; // 线程存活保持时间 private final long keepAliveTime; // 线程存活保持时间单位 private final TimeUnit unit; // 任务阻塞队列 private final BlockingQueueRunnable workQueue; // 线程工厂 private volatile ThreadFactory threadFactory; // 拒绝策略 private volatile RejectedExecutionHandler handler; // 存放线程池中的线程的集合 private final SetWorker workers new ConcurrentHashSet(); // 线程池是否安全关闭标志 private final AtomicBoolean shutDown new AtomicBoolean(false); // 线程池是否强制关闭标志 private final AtomicBoolean shutDownNow new AtomicBoolean(false); // 提交到线程池中的任务总数 private final AtomicInteger taskNum new AtomicInteger(); ...... /** * 获取线程工厂 */ public ThreadFactory getThreadFactory() { return threadFactory; } /** * 更新线程池的拒绝策略 * param handler 拒绝策略 */ public void setRejectedExecutionHandler(RejectedExecutionHandler handler) { if (handler null) { throw new NullPointerException(); } this.handler handler; } /** * 更新线程池的线程工厂 * param threadFactory 线程池 */ public void setThreadFactory(ThreadFactory threadFactory) { if (threadFactory null) { throw new NullPointerException(); } this.threadFactory threadFactory; } ...... }由于本实战仅需要实现线程池的基本简单功能因此核心线程数量最大线程数量线程保活时间和线程保活时间单位一经指定便无法再被修改。再看一下构造函数。public class ThreadPool { ...... /** * 创建线程池并使用默认的拒绝策略和线程工厂 * param coreSize 核心线程数 * param maxSize 最大线程数 * param keepAliveTime 线程保活时间 * param unit 线程保活时间单位 * param workQueue 任务阻塞队列 */ public ThreadPool(int coreSize, int maxSize, int keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue) { this.coreSize coreSize; this.maxSize maxSize; this.keepAliveTime keepAliveTime; this.unit unit; this.workQueue workQueue; threadFactory Executors.defaultThreadFactory(); handler new AbortPolicy(); } /** * 创建线程池并使用默认的拒绝策略 * param coreSize 核心线程数 * param maxSize 最大线程数 * param keepAliveTime 线程保活时间 * param unit 线程保活时间单位 * param workQueue 任务阻塞队列 * param threadFactory 线程工厂 */ public ThreadPool(int coreSize, int maxSize, int keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, ThreadFactory threadFactory) { this.coreSize coreSize; this.maxSize maxSize; this.keepAliveTime keepAliveTime; this.unit unit; this.workQueue workQueue; this.threadFactory threadFactory; handler new AbortPolicy(); } /** * 创建线程池 * param coreSize 核心线程数 * param maxSize 最大线程数 * param keepAliveTime 线程保活时间 * param unit 线程保活时间单位 * param workQueue 任务阻塞队列 * param threadFactory 线程工厂 * param handler 拒绝策略 */ public ThreadPool(int coreSize, int maxSize, int keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { this.coreSize coreSize; this.maxSize maxSize; this.keepAliveTime keepAliveTime; this.unit unit; this.workQueue workQueue; this.threadFactory threadFactory; this.handler handler; } ...... }提供了三个构造函数和ThreadPoolExecutor类似核心线程数量最大线程数量线程保活时间线程保活时间单位和任务阻塞队列需要用户指定。下面看一下submit()方法和execute()方法。public class ThreadPool { ...... /** * 执行有返回值的任务 * param callable 需要执行的任务有返回值 */ public T FutureT submit(CallableT callable) { if (callable null) { throw new NullPointerException(); } FutureTaskT futureTask new FutureTask(callable); execute(futureTask); return futureTask; } /** * 执行无返回值的任务 * param runnable 需要执行的任务无返回值 */ public void execute(Runnable runnable) { if (runnable null) { throw new NullPointerException(); } // 调用了shutDown()或者shutDownNow()方法后此时线程池不接受新任务 if (shutDown.get() || shutDownNow.get()) { return; } // 任务总数加一 taskNum.incrementAndGet(); // 如果当前线程数小于核心线程数创建Worker线程并执行任务 if (workers.size() coreSize) { addWorker(runnable); return; } // 如果当前线程数大于等于核心线程数则将任务加入任务阻塞队列 // 如果任务阻塞队列已满则offer()方法返回false表示添加失败 boolean offer workQueue.offer(runnable); if (!offer) { // 如果当前线程数小于最大线程数则创建Worker来执行任务 if (workers.size() maxSize) { addWorker(runnable); } else { // 如果当前线程数大于等于最大线程数则执行拒绝策略并且任务总数减一 taskNum.decrementAndGet(); handler.rejectedExecution(runnable, this); } } } private void addWorker(Runnable runnable) { Worker worker new Worker(runnable); workers.add(worker); worker.getThread().start(); } ...... }ThreadPool的execute()方法遵循第一小节中总结的线程池接收到一个新任务时的判断策略。addWorker()方法会创建一个Worker对象然后将其添加到Worker集合中由于使用了实现的线程安全的ConcurrentHashSet作为Worker集合因此该步骤不需要获取全局锁。Worker类是ThreadPool的内部类其实现如下。public class ThreadPool { ...... private final class Worker implements Runnable { // 要运行的初始任务创建Worker时传入的任务 private final Runnable firstTask; // 线程通过线程工厂创建 private final Thread thread; public Worker(Runnable task) { firstTask task; thread getThreadFactory().newThread(this); } public Thread getThread() { return thread; } public void close() { thread.interrupt(); } Override public void run() { // 创建Worker时会先执行初始任务 Runnable task firstTask; try { // 如果task为空则从任务阻塞队列中获取任务 while (task ! null || (task getTask()) ! null) { try { // 执行任务 task.run(); } finally { // 每次任务执行完毕会将task置为null然后循环从任务阻塞队列中获取任务 task null; taskNum.decrementAndGet(); } } } finally { // 从任务阻塞队列中获取任务为null时会跳出while循环并执行这里的释放线程逻辑 if (workers.remove(this)) { // 用于调用shutDown()方法后安全的关闭线程池 tryClose(); } } } } ...... }Worker有两个字段firstTask和thread。其中firstTask会被赋值为创建Worker时传入的任务表示Worker的初始任务因此一个Worker总是会先执行初始任务然后再去从任务阻塞队列中获取任务。thread会在创建Worker时通过线程工厂创建并且thread执行的任务为Worker自身因为Worker实现了Runnable接口Worker自身也是一个任务。在创建Worker后Worker中的线程thread便会启动从而会执行Worker的run()方法run()方法的实现思路为先执行初始任务然后循环调用ThreadPool的getTask()方法从任务阻塞队列中获取任务如果获取任务失败超过了线程保活时间/线程被中断则退出循环并执行释放Worker的逻辑。下面看一下getTask()方法的实现。public class ThreadPool { ...... private Runnable getTask() { Runnable task; // 如果强制关闭标志为true则不再从任务阻塞队列中获取任务 // 如果安全关闭标志为true且任务全部执行完则不再从任务阻塞队列中获取任务 if (shutDownNow.get() || (shutDown.get() taskNum.get() 0)) { return null; } try { // 同一时间只能有一个线程从任务阻塞队列中获取任务其余线程进入阻塞状态 lock.lockInterruptibly(); if (workers.size() coreSize) { // 如果当前线程数大于核心线程数则线程从任务阻塞队列中获取任务时最多等待线程保活的时间超时则返回null task workQueue.poll(keepAliveTime, unit); if (task null) { System.out.println(Thread.currentThread().getName() time out); } } else { // 如果当前线程数小于等于核心线程数则线程从任务阻塞队列中获取任务时会一直阻塞直到获取到任务 task workQueue.take(); } } catch (InterruptedException e) { System.out.println(Thread.currentThread().getName() interrupted); task null; } finally { lock.unlock(); } return task; } ...... }要理解getTask()方法需要结合ThreadPool的shutDown()方法和shutDownNow()方法一起分析。其实现如下。public class ThreadPool { ...... /** * 安全的关闭线程池会等待当前正在执行的任务以及任务阻塞队列中的任务执行完毕后再关闭线程池 */ public void shutDown() { shutDown.set(true); tryClose(); } /** * 强制的关闭线程池立即中断所有线程包括正在执行任务的线程 */ public void shutDownNow() { shutDownNow.set(true); doClose(); } private void tryClose() { if (shutDown.get() taskNum.get() 0) { // 如果安全关闭标志为true且任务全部执行完此时立即中断所有线程 for (Worker worker : workers) { worker.close(); } } } private void doClose() { if (shutDownNow.get()) { // 如果强制关闭标志为true此时立即中断所有线程 for (Worker worker : workers) { worker.close(); } } } }Worker通过getTask()方法从任务阻塞队列获取任务分三种情况讨论。第一种情况shutDown()方法和shutDownNow()方法没有被调用此时shutDown和shutDownNow均为false。该种情况下Worker可以正常从任务阻塞队列中获取任务由于使用了全局锁所以在同一时间只能有一个Worker能够从任务阻塞队列中获取任务其余Worker进入阻塞状态。假如线程池的当前线程数大于核心线程数那么获取到全局锁的Worker会阻塞在任务阻塞队列的poll()方法上超过保活时间还没获取到任务则直接返回null从而会释放这个Worker假如线程池的当前线程数小于等于核心线程数那么获取到全局锁的Worker会一直阻塞在任务阻塞队列的take()方法上直到获取到任务。第二种情况shutDown()方法被调用此时shutDown为true。如果还有未执行完的任务那么Worker可以正常从任务阻塞队列获取任务同情况一。如果任务阻塞队列没有任务且没有正在执行的任务那么shutDown()方法调用时便会中断所有线程在getTask()方法中阻塞的Worker会被全部唤醒并被释放掉假如shutDown()方法调用时任务阻塞队列没有任务但是存在正在执行的任务那么等待最后一个任务执行完后会中断所有线程在getTask()方法中阻塞的Worker会被全部唤醒并被释放掉。第三种情况shutDownNow()方法被调用此时shutDownNow为true。无论任务阻塞队列是否有任务以及无论是否有正在执行的任务在shutDownNow()方法调用时便会中断所有线程如果线程是阻塞在任务阻塞队列中那么线程会被唤醒并被释放掉如果线程正在执行任务那么也会尝试中断线程执行完任务或者响应中断从任务中退出的线程再次从getTask()方法获取任务时会直接获取到null因为shutDownNow为true从而被释放掉。特别注意如果某个线程执行的任务是一个死循环并且无法响应中断那么这个线程永远不会被释放。最后ThreadPool提供一个size()方法用于获取当前的Worker数量该方法仅用于测试。public int size() { return workers.size(); }下面编写测试程序来测试ThreadPool。测试用例一测试ThreadPool的shutDown()方法创建一个ThreadPool核心线程数为2最大线程数为4任务阻塞队列大小为20创建一个固定任务为打印数字0-2每打印一次数字睡眠1秒。首先向ThreadPool提交三个固定任务等待4秒后执行shutDown()方法预期的结果应该为提交的三个任务都会执行完并且任务执行完后阻塞在任务阻塞队列中的任务会被唤醒并被释放。测试代码如下。public class ThreadPoolTest { Test void testThreadPool_1() { CountDownLatch countDownLatch new CountDownLatch(3); BlockingQueueRunnable workQueue new ArrayBlockingQueue(20); ThreadPool threadPool new ThreadPool(2, 4, 1000, TimeUnit.MILLISECONDS, workQueue); Runnable task () - { try { for (int i 0; i 3; i) { System.out.println(Thread.currentThread().getName() : i); Thread.sleep(1000); } } catch (InterruptedException e) { System.out.println(Thread.currentThread().getName() interrupted from sleep); } countDownLatch.countDown(); }; for (int i 0; i 3; i) { threadPool.execute(task); } try { Thread.sleep(4000); threadPool.shutDown(); Thread.sleep(3000); countDownLatch.await(); } catch (InterruptedException e) { System.out.println(e.getMessage()); } System.out.println(Worker size: threadPool.size()); System.out.println(Task num: workQueue.size()); } }测试结果如下所示。结果表明一开始ThreadPool分别创建了线程1和线程2来处理task1和task2然后task3被添加到了任务阻塞队列当线程1和线程2执行完任务后线程2获取到了全局锁从任务阻塞队列中获取到了task3而线程1进入阻塞状态此时调用shutDown()方法等待线程2执行完任务后中断所有线程此时线程1被唤醒然后被释放掉最后ThreadPool中线程数量和任务数量全部为0。测试用例二测试ThreadPool的shutDownNow()方法创建一个ThreadPool核心线程数为2最大线程数为4任务阻塞队列大小为20创建一个固定任务为打印数字0-2每打印一次数字睡眠1秒。首先向ThreadPool提交三个固定任务等待2秒后执行shutDownNow()方法预期的结果应该为提交的三个任务都不会执行完所有线程在shutDownNow()方法被调用后会被中断并释放掉。测试代码如下。public class ThreadPoolTest { Test void testThreadPool_2() { CountDownLatch countDownLatch new CountDownLatch(2); BlockingQueueRunnable workQueue new ArrayBlockingQueue(20); ThreadPool threadPool new ThreadPool(2, 4, 1000, TimeUnit.MILLISECONDS, workQueue); Runnable task () - { try { for (int i 0; i 3; i) { System.out.println(Thread.currentThread().getName() : i); Thread.sleep(1000); } } catch (InterruptedException e) { System.out.println(Thread.currentThread().getName() interrupted from sleep); } countDownLatch.countDown(); }; for (int i 0; i 3; i) { threadPool.execute(task); } try { Thread.sleep(2000); threadPool.shutDownNow(); Thread.sleep(1000); countDownLatch.await(); } catch (InterruptedException e) { System.out.println(e.getMessage()); } System.out.println(Worker size: threadPool.size()); System.out.println(Task num: workQueue.size()); } }测试结果如下所示。结果表明一开始ThreadPool分别创建了线程1和线程2来处理task1和task2然后task3被添加到了任务阻塞队列然后在线程1和线程2执行任务的过程中调用了shutDownNow()方法由于task1和task2能够响应中断并退出任务所以线程1和线程2会从任务中退出然后被释放掉任务阻塞队列中的task3也不会被执行最后ThreadPool中线程数量为0任务数量为1。测试用例三创建一个ThreadPool核心线程数为2最大线程数为4任务阻塞队列大小为1线程保活时间为1秒创建一个固定任务为打印数字0-2每打印一次数字睡眠1秒。首先向ThreadPool提交五个固定任务然后等待ThreadPool执行这些任务预期的结果应该为在所有任务执行完后ThreadPool中的线程数应该为2核心线程数任务数应该为0。测试代码如下。public class ThreadPoolTest { Test void testThreadPool_3() { CountDownLatch countDownLatch new CountDownLatch(5); BlockingQueueRunnable workQueue new ArrayBlockingQueue(1); ThreadPool threadPool new ThreadPool(2, 4, 1000, TimeUnit.MILLISECONDS, workQueue); Runnable task () - { try { for (int i 0; i 3; i) { System.out.println(Thread.currentThread().getName() : i); Thread.sleep(1000); } } catch (InterruptedException e) { System.out.println(Thread.currentThread().getName() interrupted from sleep); } countDownLatch.countDown(); }; for (int i 0; i 5; i) { threadPool.execute(task); } try { countDownLatch.await(); } catch (InterruptedException e) { System.out.println(e.getMessage()); } System.out.println(Worker size: threadPool.size()); System.out.println(Task num: workQueue.size()); } }测试结果如下所示。结果表明一开始ThreadPool创建了线程1和线程2来处理task1和task2然后task3被添加到了任务阻塞队列然后创建了线程3和线程4来处理task4和task5在任务执行完后线程4获取到了全局锁从任务阻塞队列中获取到了task3继续执行此时线程1和线程2由于在线程保活时间内没有获取到任务来执行被释放掉最后ThreadPool中线程数量为2任务数量为0。