微交易系统开发语言之Java线程池

微交易系统开发学习之 Java 线程池 线程池的技术背景线程池的技术背景 在面向对象编程中 创建和销毁对象是很费时间的 因为创建一个对象要获取内存资 源或者其它更多资源 在 Java 中更是如此 虚拟机将试图跟踪每一个对象 以便能够在对 象销毁后进行垃圾回收 所以提高服务程序效率的一个手段就是尽可能减少创建和销毁对象的次数 特别是一 些很耗资源的对象创建和销毁 如何利用已有对象来服务就是一个需要解决的关键问题 其实这就是一些 池化资源 技术产生的原因 例如 Android 中常见到的很多通用组件一般都离不开 池 的概念 如各种图片加载 库 网络请求库 即使 Android 的消息传递机制中的 Meaasge 当使用 Meaasge obtain 就 是使用的 Meaasge 池中的对象 因此这个概念很重要 本文将介绍的线程池技术同样符合 这一思想 线程池的优点线程池的优点 重用线程池中的线程 减少因对象创建 销毁所带来的性能开销 能有效的控制线程的最 大并发数 提高系统资源利用率 同时避免过多的资源竞争 避免堵塞 能够多线程进行简单 的管理 使线程的使用简单 高效 线程池框架线程池框架 ExecutorExecutor java 中的线程池是通过 Executor 框架实现的 Executor 框架包括类 Executor Executors ExecutorService ThreadPoolExecutor Callable 和 Future FutureTask 的使用等 ExecutorExecutor 所有线程池的接口 只有一个方法 1 2 3 public interface Executor void execute Runnable command ExecutorServiceExecutorService 增加 Executor 的行为 是 Executor 实现类的最直接接口 ExecutorsExecutors 提供了一系列工厂方法用于创先线程池 返回的线程池都实现了 ExecutorService 接口 ThreadPoolExecutorThreadPoolExecutor 线程池的具体实现类 一般用的各种线程池都是基于这个类实现的 构造方法如下 1 2 3 4 5 6 public ThreadPoolExecutor int corePoolSize int maximumPoolSize long keepAliveTime TimeUnit unit BlockingQueue workQueue 7 8 9 10 this corePoolSize maximumPoolSize keepAliveTime unit workQueue Executors defaultThreadFactory defaultHandler corePoolSizecorePoolSize 线程池的核心线程数 线程池中运行的线程数也永远不会超过 corePoolSize 个 默认情况下可以一直存活 可以通过设置 allowCoreThreadTimeOut 为 True 此时 核心线程数就是核心线程数就是 0 0 此时 keepAliveTime 控制所有线程的超时时间 maximumPoolSizemaximumPoolSize 线程池允许的最大线程数 keepAliveTimekeepAliveTime 指的是空闲线程结束的超时时间 unitunit 是一个枚举 表示 keepAliveTime 的单位 workQueueworkQueue 表示存放任务的 BlockingQueue Runnable 队列 BlockingQueueBlockingQueue 阻塞队列 BlockingQueue 是 java util concurrent 下的主要用来控制 线程同步的工具 如果 BlockQueue 是空的 从 BlockingQueue 取东西的操作将会被阻断进 入等待状态 直到 BlockingQueue 进了东西才会被唤醒 同样 如果 BlockingQueue 是满的 任何试图往里存东西的操作也会被阻断进入等待状态 直到 BlockingQueue 里有空间才会被 唤醒继续操作 阻塞队列常用于生产者和消费者的场景 生产者是往队列里添加元素的线程 消费者是从 队列里拿元素的线程 阻塞队列就是生产者存放元素的容器 而消费者也只从容器里拿元 素 具体的实现类有 LinkedBlockingQueue ArrayBlockingQueued 等 一般其内部的都是 通过 Lock 和 Condition 显示锁 Lock 及 Condition 的学习与使用 来实现阻塞和唤醒 线程池的工作过程如下线程池的工作过程如下 1 1 线程池刚创建时 里面没有一个线程 任务队列是作为参数传进来的 不过 就算队列 里面有任务 线程池也不会马上执行它们 2 当调用 execute 方法添加一个任务时 线程池会做如下判断 3 如果正在运行的线程数量小于 corePoolSize 那么马上创建线程运行这个任务 如果正在运行的线程数量大于或等于 corePoolSize 那么将这个任务放入队列 如果这时候队列满了 而且正在运行的线程数量小于 maximumPoolSize 那么还是要创建 非核心线程立刻运行这个任务 如果队列满了 而且正在运行的线程数量大于或等于 maximumPoolSize 那么线程池会抛 出异常 RejectExecutionException 4 当一个线程完成任务时 它会从队列中取下一个任务来执行 5 当一个线程无事可做 超过一定的时间 keepAliveTime 时 线程池会判断 如果当前 运行的线程数大于 corePoolSize 那么这个线程就被停掉 所以线程池的所有任务完成后 6 它最终会收缩到 corePoolSize 的大小 线程池的创建和使用线程池的创建和使用 生成线程池采用了工具类 Executors 的静态方法 以下是几种常见的线程池 SingleThreadExecutorSingleThreadExecutor 单个后台线程 其缓冲队列是无界的 1 2 3 4 5 6 public static ExecutorService newSingleThreadExecutor return new FinalizableDelegatedExecutorService new ThreadPoolExecutor 1 1 0L TimeUnit MILLISECONDS new LinkedBlockingQueue 创建一个单线程的线程池 这个线程池只有一个核心线程在工作 也就是相当于单线程串 行执行所有任务 如果这个唯一的线程因为异常结束 那么会有一个新的线程来替代它 此线程池保证所有任务的执行顺序按照任务的提交顺序执行 FixedThreadPoolFixedThreadPool 只有核心线程的线程池 大小固定 其缓冲队列是无界的 1 2 3 4 5 public static ExecutorService newFixedThreadPool int nThreads return new ThreadPoolExecutor nThreads nThreads 0L TimeUnit MILLISECONDS new LinkedBlockingQueue 创建固定大小的线程池 每次提交一个任务就创建一个线程 直到线程达到线程池的最大 大小 线程池的大小一旦达到最大值就会保持不变 如果某个线程因为执行异常而结束 那么线程池会补充一个新线程 CachedThreadPoolCachedThreadPool 无界线程池 可以进行自动线程回收 1 2 3 4 5 public static ExecutorService newCachedThreadPool return new ThreadPoolExecutor 0 Integer MAX VALUE 60L TimeUnit SECONDS new SynchronousQueue 如果线程池的大小超过了处理任务所需要的线程 那么就会回收部分空闲 60 秒不执行任 务 的线程 当任务数增加时 此线程池又可以智能的添加新线程来处理任务 此线程池 不会对线程池大小做限制 线程池大小完全依赖于操作系统 或者说 JVM 能够创建的最 大线程大小 SynchronousQueue 是一个是缓冲区为 1 的阻塞队列 ScheduledThreadPool 核心线程池固定 大小无限的线程池 此线程池支持定时以及周期 性执行任务的需求 1 2 3 4 5 6 public static ExecutorService newScheduledThreadPool int corePoolSize return new ScheduledThreadPool corePoolSize Integer MAX VALUE DEFAULT KEEPALIVE MILLIS MILLISECONDS new DelayedWorkQueue 创建一个周期性执行任务的线程池 如果闲置 非核心线程池会在 DEFAULT KEEPALIVEMILLIS 时间内回收 线程池最常用的提交任务的方法有两种 executeexecute 1ExecutorService execute Runnable runable submitsubmit 1 2 3 4 5 FutureTask task ExecutorService submit Runnable runnable FutureTask task ExecutorService submit Runnable runnable T Result FutureTask task ExecutorService submit Callable callable submit Callable callable 的实现 submit Runnable runnable 同理 1 2 public Future submit Callable task if task null throw new NullPointerException 3 4 5 6 FutureTask ftask newTaskFor task execute ftask return ftask 可以看出 submit 开启的是有返回结果的任务 会返回一个 FutureTask 对象 这样就能通 过 get 方法得到结果 submit 最终调用的也是 execute Runnable runable submit 只 是将 Callable 对象或 Runnable 封装成一个 FutureTask 对象 因为 FutureTask 是个 Runnable 所以可以在 execute 中执行 关于 Callable 对象和 Runnable 怎么封装成 FutureTask 对象 见 Callable 和 Future FutureTask 的使用 线程池实现的原理线程池实现的原理 如果只讲线程池的使用 那这篇博客没有什么大的价值 充其量也就是熟悉 Executor 相关 API 的过程 线程池的实现过程没有用到 Synchronized 关键字 用的都是 volatile Lock 和同步 阻塞 队列 Atomic 相关类 FutureTask 等等 因为后者的性能更优 理解的过程 可以很好的学习源码中并发控制的思想 在开篇提到过线程池的优点是可总结为以下三点 1 线程复用 2 控制最大并发数 3 管理线程 1 1 线程复用过程线程复用过程 理解线程复用原理首先应了解线程生命周期 在线程的生命周期中 它要经过新建 New 就绪 Runnable 运行 Running 阻塞 Blocked 和死亡 Dead 5 种状态 Thread 通过 new 来新建一个线程 这个过程是是初始化一些线程信息 如线程名 id 线 程所属 group 等 可以认为只是个普通的对象 调用 Thread 的 start 后 Java 虚拟机会 为其创建方法调用栈和程序计数器 同时将 hasBeenStarted 为 true 之后调用 start 方法 就会有异常 处于这个状态中的线程并没有开始运行 只是表示该线程可以运行了 至于该线程何时开 始运行 取决于 JVM 里线程调度器的调度 当线程获取 cpu 后 run 方法会被调用 不要 自己去调用 Thread 的 run 方法 之后根据 CPU 的调度在就绪 运行 阻塞间切换 直到 run 方法结束或其他方式停止线程 进入 dead 状态 所以实现线程复用的原理应该就是要保持线程处于存活状态 就绪 运行或阻塞 接下 来来看下 ThreadPoolExecutor 是怎么实现线程复用的 在 ThreadPoolExecutor 主要 Worker 类来控制线程的复用 看下 Worker 类简化后的代码 这样方便理解 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 private final class Worker implements Runnable final Thread thread Runnable firstTask Worker Runnable firstTask this firstTask firstTask this thread getThreadFactory newThread this public void run runWorker this final void runWorker Worker w Runnable task w firstTask w firstTask null while task null task getTask null 20 21 22 23 task run Worker 是一个 Runnable 同时拥有一个 thread 这个 thread 就是要开启的线程 在新建 Worker 对象时同时新建一个 Thread 对象 同时将 Worker 自己作为参数传入 TThread 这 样当 Thread 的 start 方法调用时 运行的实际上是 Worker 的 run 方法 接着到 runWorker 中 有个 while 循环 一直从 getTask 里得到 Runnable 对象 顺序执行 getTask 又是怎么得到 Runnable 对象的呢 依旧是简化后的代码 private Runnable getTask if 一些特殊情况 return null Runnable r workQueue take return r 这个 workQueue 就是初始化 ThreadPoolExecutor 时存放任务的 BlockingQueue 队列 这个 队列里的存放的都是将要执行的 Runnable 任务 因为 BlockingQueue 是个阻塞队列 BlockingQueue take 得到如果是空 则进入等待状态直到 BlockingQueue 有新的对象被 加入时唤醒阻塞的线程 所以一般情况 Thread 的 run 方法就不会结束 而是不断执行从 workQueue 里的 Runnable 任务 这就达到了线程复用的原理了 2 2 控制最大并发数控制最大并发数 那 Runnable 是什么时候放入 workQueue Worker 又是什么时候创建 Worker 里的 Thread 的又是什么时候调用 start 开启新线程来执行 Worker 的 run 方法的呢 有上面的分析 看出 Worker 里的 runWorker 执行任务时是一个接一个 串行进行的 那并发是怎么体现 的呢 很容易想到是在 execute Runnable runnable 时会做上面的一些任务 看下 execute 里是 怎么做的 execute execute 简化后的代码 public void execute Runnable command if command null throw new NullPointerException int c ctl get 当前线程数 corePoolSize if workerCountOf c corePoolSize runState 为 RUNNING 再次检验是否为 RUNNING 状态 非 RUNNING 状态 则从 workQueue 中移除任务并拒绝 if isRunning recheck 采用线程池指定的策略拒绝任务 两种情况 1 非 RUNNING 状态拒绝新的任务 2 队列满了启动新的线程失败 workCount maximumPoolSize else if addWorker command false reject command addWorker addWorker 简化后的代码 private boolean addWorker Runnable firstTask boolean core int wc workerCountOf c if wc core corePoolSize maximumPoolSize return false w new Worker firstTask final Thread t w thread t start 根据代码再来看上面提到的线程池工作过程中的添加任务的情况 如果正在运行的线程数量小于 corePoolSize 那么马上创建线程运行这个任务 如果正在运行的线程数量大于或等于 corePoolSize 那么将这个任务放入队列 如果这时候队列满了 而且正在运行的线程数量小于 maximumPoolSize 那么还是 要创建非核心线程立刻运行这个任务 如果队列满了 而且正在运行的线程数量大于或等于 maximumPoolSize 那么线程 池会抛出异常 RejectExecutionException 这就是 Android 的 AsyncTask 在并行执行是在超出最大任务数是抛出 RejectExecutionException 的原因所在 详见基于最新版本的 AsyncTask 源码解读及 AsyncTask 的黑暗面 通过 addWorker 如果成功创建新的线程成功 则通过 start 开启新线程 同时将 firstTask 作为这个 Worker 里的 run 中执行的第一个任务 虽然每个 Worker 的任务是串行处理 但如果创建了多个 Worker 因为共用一个 workQueue 所以就会并行处理了 所以根据 corePoolSize 和 maximumPoolSize 来控制最大并发数 大致过程可用下图表示 上面的讲解和图来可以很好的理解的这个过程 如果是做 Android 开发的 并且对 Handler 原理比较熟悉 你可能会觉得这个图挺熟悉 其中的一些过程和 Handler Looper Meaasge 使用中 很相似 Handler send Message 相当于 execute Runnuble Looper 中维护的 Meaasge 队列相当于 BlockingQueue 只不 过需要自己通过同步来维护这个队列 Looper 中的 loop 函数循环从 Meaasge 队列取 Meaasge 和 Worker 中的 runWork 不断从 BlockingQueue 取 Runnable 是同样的道理 3 3 管理线程管理线程 通过线程池可以很好的管理线程的复用 控制并发数 以及销毁等过程 线程的复用和控制 并发上面已经讲了 而线程的管理过程已经穿插在其中了