B2. Concurrent 线程池（Executor）

【概述】

　　与数据库连接管理类似，线程的创建和销毁会耗费较大的开销，使用 “池化技术” 来更好地利用当前线程资源，减少因线程创建和销毁带来的开销，这就是线程池产生的原因。

【无限创建线程的不足】

　　在生产环境中，若没有线程池，则需要采用的是 “为每个任务创建一个线程” 的方法，当出现大量的请求时需要创建大量的线程：

线程生命周期的开销非常高：线程的创建和销毁并不是没有代价的。根据平台的不同，实际的开销也有所不同，但线程的创建过程都需要时间，延迟处理的请求，并且需要 JVM 和操作系统提供一些辅助操作。如果请求的到达率非常高且请求的处理过程是轻量级的，例如大多数服务器应用程序就是这种情况，那么为每个请求创建一个新线程将消耗大量的计算资源。
资源消耗：活跃的线程会消耗系统资源，尤其是内存。如果可运行的线程数量多于可用处理器的数量，那么有些线程将闲置。大量空闲的线程会占用许多内存，给垃圾回收器带来压力，而且大量线程在竞争 CPU 资源时还将产生其他的性能开销。如果你已经拥有足够多的线程使 CPU 保持忙碌状态，那么再创建更多的线程反而会降低性能。
稳定性：在可创建线程的数量上存在一个限制。这个限制值将随着平台的不同而不同，并且受多个因素制约，包括 JVM 的启动参数、Thread 构造函数中请求的栈大小，以及底层操作系统对线程的限制等。如果破坏了这些限制，那么很可能会抛出 OOM 异常，要想从这种错误中恢复过来是非常危险的，更简单的方法是通过构造程序来避免超出这些限制。

　　在一定的范围内，增加线程可以提供系统的吞吐率，但如果超出了这个范围，再创建再多的线程只会降低程序的执行速度，并且如果过多地创建一个线程，那么整个应用程序将会崩溃。要想避免这种危险，就应该对应用程序可以创建的线程数量进行限制，并且全面地测试应用程序，从而确保在线程数量达到限制时，程序也也不会耗尽资源。

　　“为每个任务分配一个线程” 这种方法的问题在于，它没有限制可创建线程的数量，只限制了远程用户提交 HTTP 请求的速率。与其他的并发危险一样，在原型设计和开发阶段，无限制地创建线程或许还能较好地运行，但在应用程序部署后并处于高负载下运行时，才会有问题不断地暴露出来。因此，某个恶意的用户或者过多的用户，都会使 Web 服务器的负载达到某个阈值，从而使服务器崩溃。如果服务器需要提供高可用性，并且在高负载情况下能平缓地降低性能，那么这将是一个严重的故障。

【线程池的执行策略】

　　1). 在执行（execute）一个任务（command）时，获取当前线程池状态（c），然后获取线程池工作线程的数量（workerCountOf(c)），如果当前线程池的工作线程数量少于定义的核心工作线程的数量（corePoolSize），则使用工厂（threadFactory）创建一个核心工作线程；若核心工作线程已满，则无需再创建核心工作线程，继续下一步。

　　2). 这里对线程池状态进行进行双检查（double-check）。首先进行第一次状态检查，获取线程池的状态（ctl），判断其是否处于运行状态（isRunning）；

　　2.1). 如果线程池处于运行状态（isRunning 方法返回结果为 true），则将任务加入到工作队列中（workQueue，是一个阻塞队列 BlockingQueue）；如果成功加入（offer 方法返回 true）到工作队列（workQueue），此时进行第二次状态检查，获取线程池的状态（recheck），判断其是否处于运行状态（isRunning 方法返回结果为 true）；

　　　　2.1.1). 如果线程池处于运行状态，则表示该任务可以被执行，此时判断当前工作线程数量是否为空（workerCountOf(recheck) 返回 0），若为空则使用工厂（threadFactory）创建一个非核心工作线程（总的工作线程数量不能大于最大工作线程数量 maximumPoolSize）；

　　　　2.1.2). 如果线程池处于非运行状态，则表示该任务不可以被执行，此时需要从工作队列中移除（remove）该任务，若移除任务失败，则需要交给拒绝执行处理器（handler）进行拒绝处理（reject）。

　　2.2). 如果线程池处于非运行状态（isRunning 方法返回结果为 false）或加入工作队列（workQueue）失败（offer 方法返回 false），则使用工厂（threadFactory）创建一个非核心工作线程（总的工作线程数量不能大于最大工作线程数量 maximumPoolSize）；如果创建失败，则需要交给拒绝执行处理器（handler）进行拒绝处理（reject）。

　　JDK 中实现代码如下：

    public void execute(Runnable command) {

        if (command == null)

            throw new NullPointerException();

        /*

         * Proceed in 3 steps:

         *

         * 1. If fewer than corePoolSize threads are running, try to

         * start a new thread with the given command as its first

         * task.  The call to addWorker atomically checks runState and

         * workerCount, and so prevents false alarms that would add

         * threads when it shouldn't, by returning false.

         *

         * 2. If a task can be successfully queued, then we still need

         * to double-check whether we should have added a thread

         * (because existing ones died since last checking) or that

         * the pool shut down since entry into this method. So we

         * recheck state and if necessary roll back the enqueuing if

         * stopped, or start a new thread if there are none.

         *

         * 3. If we cannot queue task, then we try to add a new

         * thread.  If it fails, we know we are shut down or saturated

         * and so reject the task.

         */

        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);

    }

巴特西

B2. Concurrent 线程池（Executor）

最新文章

热门文章