ThreadPoolExecutor的配置（二）

配置ThreadPoolExecutor二

线程的建立和销毁

线程池的基本大小（Core Pool Size）、最大大小（Maximum Pool Size）以及存活时间（keepAliveTime）等因素共同负责线程的建立和销毁。基本大小也就是线程池的目标大小，即在没有任务执行时线程池的大小，而且只有在工做队列满了的状况下才会建立超出这个数量的线程。线程池的最大大小表示可同时活动的线程数量的上限。若是某个线程的空闲时间超过了存活时间，那么将被标记为可回收的，而且当线程池的当前大小超过了基本大小时，这个线程将被终止。

经过调节线程池的基本大小和存活时间，能够帮助线程池回收空闲线程占有的资源，从而使得这些资源能够用于执行其余工做。（显然，这是一种折衷：回收空闲线程会产生额外的延迟，由于当需求增长时，必须建立新的线程来知足需求。）

newFixedThreadPool工厂方法将线程池的基本大小和最大大小设置为参数中指定的值，

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(nThreads, nThreads,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(),
            threadFactory);
}

newCachedThreadPool工厂方法将线程池的最大大小设置为Integer.MAX_VALUE，而将基本大小设置为零，并将超时设置为1分钟，

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
            60L, TimeUnit.SECONDS,
            new SynchronousQueue<Runnable>());
}

这种方法建立出来的线程池能够被无限扩展，而且当需求下降时会自动收缩。其余形式的线程池能够经过显式的ThreadPoolExecutor 构造函数来构造。

注：在建立ThreadPoolExecutor的初期，线程并不会当即启动，而是等到有任务提交时才会启动，除非调用prestartAllCoreThreads

管理队列任务

在有限的线程池中会限制可并发执行的任务数量。若是无限制的建立线程，那么将致使不稳定性，并经过采用固定大小的线程池来解决这个问题。然而这个方案并不完整。在高负载状况下，应用程序仍可能耗尽资源，只是出现问题的几率较小。若是新请求的到达速率超过了线程池的处理速率，那么新到来的请求将累积起来。在线程池中，这些请求会在一个由Executor管理的Runnable队列中等待，而不会像线程那样去竞争CPU资源。

即便请求的平均到达速率很稳定，也仍然会出现请求突增的状况。尽管队列有助于缓解任务的突增问题，但若是任务持续高速的到来，那么最终仍是会抑制请求的到达速率以免耗尽内存。

ThreadPoolExecutor容许提供一个BlockingQueue来保存等待执行的任务。基本的任务排队方法有三种：无界队列、有界队列和同步移交（Synchronous Handleroff）。队列的选择和其余的配置参数有关，例如线程池的大小。

newFixedThreadPool 和 newSingleThreadExecutor 在默认状况下将使用一个无界的LinkedBlockingQueue。若是全部工做者线程都处于忙碌状态，那么任务将在队列中等候。若是任务持续快速的到达，而且超过了线程池的处理速度，那么队列将无限增长。

一种更稳妥的资源管理策略是使用有界队列，例如ArrayBlockingQueue、 有界的LinkedBlockingQueue、PriorityBlockingQueue（PriorityBlockingQueue 能够指定初始的队列大小，后面插入元素的时候，若是空间不够的话会自动扩容）。有界队列有助于避免资源耗尽的状况发生，但他又带来了新的问题：当队列填满后，新的任务该怎么办？

在使用有界的工做队列时，队列的大小与线程池的大小必须一块儿调节。若是线程池较小而队列较大，那么有助于减小内存使用量，下降CPU的使用率，同时还能够减小上下文切换，但付出的代价是可能会限制吞吐量。

**对于很是大的或者无界的线程池来讲，能够经过SynchronousQueue来避免任务排队，以及直接将任务从生产者移交给工做者线程。**SynchronousQueue不是一个真正的队列，而是一种在线程之间进行移交的机制。要将一个元素放入SynchronousQueue中，必须有另外一个线程正在等待接受这个元素。若是没有线程正在等待，而且线程池的当前大小小于最大值，那么ThreadPoolExecutor将建立一个新的线程，不然根据饱和策略，这个任务将被拒绝。使用直接移交将更高效，由于任务会直接移交给执行它的线程，而不是放在队列中，而后由工做者线程从队列中提取该任务。只有当线程池是无界的或者能够拒绝任务时，SynchronousQueue才有实际价值。

当使用像LinkedBlockingQueue或ArrayBlockingQueue这样的FIFO队列时，任务的执行顺序与他们的到达顺序相同。若是像进一步控制任务的执行顺序，还可使用PriorityBlockingQueue，这个队列将根据优先级来安排任务。任务的优先级时经过天然顺序或者Compator来定义的。

饱和策略

当有界队列被填满后，饱和策略开始发挥做用。ThreadPoolExecutor的饱和策略能够经过调用setRejectedExecutionHandler 来修改。JDK提供了几种不一样的RejectedExecutionHandler 实现，每种实现包含了不一样的饱和策略：

AbortPolicy、CallerRunsPolicy、DiscardPolicy、DiscardOldestPolicy

停止策略：是默认的饱和策略，该策略将抛出未检查的RejectedExecutionException异常，调用者能够捕获这个异常，而后根据需求编写本身的处理代码。

抛弃策略：当新提交的任务没法保存到队列中等待执行是，抛弃策略会悄悄抛弃该任务。

抛弃最旧策略：将会抛弃下一个将被执行的任务，而后尝试从新提交新的任务（若是工做队列是一个优先队列，那么抛弃最旧的策略将致使抛弃优先级最高的任务，所以最好不要将抛弃最旧的饱和策略和优先级队列放在一块儿使用）。

调用者运行（Caller-Runs）策略：实现了一种调节机制，该策略既不会抛弃任务，也不会抛出异常，而是将某些任务回退到调用者，从而下降新任务的流量。他不会在线程池的某个线程中执行，而是在一个调用了execute的线程中执行该任务。

线程工厂

每当线程池须要建立一个线程时，都是经过线程工厂方法来完成的。默认的线程工厂将建立一个新的、非守护的线程，而且包含特殊的配置信息。经过指定一个线程工厂方法，能够定制线程池的配置信息。在ThreadFactory中只定义一个方法newThread，每当线程池须要建立一个新线程时都会调用这个方法。

使用 Semaphore 来控制任务的提交速率

import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;

public class BoundedExecutor {

    private final Executor exec;
    private final Semaphore semaphore;

    public BoundedExecutor(Semaphore semaphore, Executor exec) {
        this.semaphore = semaphore;
        this.exec = exec;
    }

    public void submit(final Runnable command) throws InterruptedException {
        semaphore.acquire();
        try {
            exec.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        command.run();
                    } finally {
                        semaphore.release();
                    }
                }
            });

        } catch (RejectedExecutionException e) {
            semaphore.release();
        }
    }
}

=========END=========