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=========