线程池ThreadPoolExecutor应用过程

一个老生常谈的话题：线程的创建及销毁是非常消耗时间及资源的，所以线程应该交由线程池去执行。

ThreadPoolExecutor构造说明及常用方法

ThreadPoolExecutor提供了很多构造方法，这里主要说以下这个，其他构造方法都是基于此方法：

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

• 参数int corePoolSize：核心线程池大小，不能小于0；
• 参数int maximumPoolSize：最大线程池大小(核心线程数+临时线程数)，不能小于等于0且不能小于corePoolSize参数，线程池的大小会在corePoolSize与maximumPoolSize之间动态变化；
• 参数long keepAliveTime：临时线程(非核心线程，核心线程空闲之后不会被销毁)最大空闲时间，不能小于0；
• 参数TimeUnit unit：keepAliveTime参数的时间单位；
• 参数BlockingQueue<Runnable> workQueue：线程等待队列；
• 参数ThreadFactory threadFactory：线程创建工厂；
• 参数RejectedExecutionHandler handler：拒绝策略；

下面是构造ThreadPoolExecutor的参数校验：

其他常用方法：

• int prestartAllCoreThreads()：根据线程核心数，启动所有的核心线程，让它们全部处于空闲状态等待工作
• boolean prestartCoreThread()：随机启动一个核心线程，当所有核心线程都已经启动的时候，返回false(表示未启动到有效资源)；
• void allowCoreThreadTimeOut(boolean value)：设置是否允许临时线程永久空闲存活，设置true后，keepAliveTime参数失效；
• boolean allowsCoreThreadTimeOut()：获取是否允许临时线程永久空闲存活；
• void shutdown()：销毁线程池，此前还存在于线程池的任务依然会被执行；
• List<Runnable> shutdownNow()：销毁线程池，与shutdown()的区别是，未执行完成的线程会被标记为中断；

参数举例说明：

1、当创建一个任务并交给线程池执行执行后，会从消耗一个核心线程资源，当该任务执行完毕之后，会变成一个空闲线程，另外，如果核心线程数足以支撑任务运行，即使有核心空闲线程，依然会创建一个新的核心线程资源；

2、当加入线程池的任务超过核心线程数，会进入workQueue中等待执行；

3、当加入线程池的任务超过核心线程数，溢出的任务数如果超过等待队列长度，会直接创建大于corePoolSize且小于maximumPoolSize的线程资源用于执行任务，超过核心线程数的部分称为临时线程；

4、keepAliveTime：当线程池被扩大到corePoolSize与maximumPoolSize之间之后，当执行完所有任务后，不再有新的任务进来，超出corePoolSize的部分会作为空闲线程，会在keepAliveTime指定时间后，线程池大小缩小为corePoolSize，如果设置allowCoreThreadTimeOut(true),则空闲线程一直不会销毁；

5、BlockingQueue<Runnable> workQueue：用于保存等待执行的任务的阻塞队列接口，java提供了以下实现类：

线程池中，BlockingQueue是典型的"生产者消费者"模型：生产者调用插入元素的方法(add/offer/put等方法)，消费者调用移除元素的方法(remove/poll/take等方法)，其中，生产者插入和消费者移除都有阻塞和非阻塞的方法，offer/poll是非阻塞方法，put/take是阻塞方法(队列后面单独再写)；

ThreadPoolExecutor提交任务采用的是不阻塞的offer方法，从队列中获取任务采用的是阻塞的take方法，正是因为ThreadPoolExecutor使用了不阻塞的offer方法，所以当队列容量已满，线程池会去创建新的临时线程，去处理队列中溢出的任务。

• ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序，允许迭代中修改队列(删除元素时会更新迭代器)，当然，一般也不会有场景需要迭代阻塞队列；
• LinkedBlockingQueue：一个基于链表结构的默认无界双端阻塞队列，双端意味着可以像普通队列一样FIFO(先进先出)，可以以像栈一样FILO(先进后出)。吞吐量通常要高于ArrayBlockingQueue(如无需在迭代中修改元素，优于ArrayBlockingQueue，同时记得指定队列长度，提供了一个可选有界的构造函数，而在未指明容量时，容量默认为Integer.MAX_VALUE)；
• SynchronousQueue：一个不存储元素的阻塞队列，所谓不存储元素，正如名字描述，同步队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，即生产一个，就消费一个，因此它也被认为是无界的队列，吞吐量通常要高于LinkedBlockingQueue。同时，SynchronousQueue不能遍历，因为它没有元素可以遍历；
• DelayQueue：使用二叉堆实现的优先级阻塞队列，通过执行时延从队列中提取任务，时间没到任务取不出来，可用于实现有延时场景的需求；
• PriorityBlockingQueue：使用二叉堆实现的优先级阻塞队列，可以实现Comparable接口也可以提供Comparator来对队列中的元素进行比较，跟时间没有任何关系，仅仅是按照优先级取任务，如果有根据某个规则优先取出任务的场景，适用；

综上，需要根据每个队列的特点选择适合自己场景的队列。

6、RejectedExecutionHandler handler：当队列和线程池都满了(从上面可以看出，一个线程池可以容纳的的最大任务数是maximumPoolSize+队列长度，因此使用无界队列会造成队列永不会满的情况，一旦控制不好就容易出现OOM)，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。

1、AbortPolicy：直接抛出异常；

2、CallerRunsPolicy：只用调用者所在线程来运行任务；

3、DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务；

4、DiscardPolicy：不处理，丢弃掉；

5、也可以根据应用场景需要来实现RejectedExecutionHandler接口自定义策略。如记录日志或持久化不能处理的任务；

注：AbortPolicy/CallerRunsPolicy/DiscardOldestPolicy/DiscardPolicy都是ThreadPoolExecutor的内部类，如下：

所以，实例化需要用内部类的方式，如下：

AbortPolicy abortPolicy = new ThreadPoolExecutor.AbortPolicy();

自定义RejectedExecutionHandler拒绝策略示例：

public class RejectedExecutionHandlerTest implements RejectedExecutionHandler {

	@Override
	public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
		// 啥逻辑都没有,表示忽略,比如DiscardPolicy
		try {
			executor.getQueue().put(r);
		} catch (InterruptedException e) {
			// 	TODO 记录错误日志等
			e.printStackTrace();
		}
		
	}
}

ThreadPoolExecutor的特点总结起来就是以下4点：

1、当有任务提交的时候，会创建核心线程去执行任务（即使有核心线程空闲）；

2、当核心线程数达到corePoolSize时，后续提交的都会进BlockingQueue中排队；

3、当BlockingQueue满了(offer失败)，就会创建临时线程(临时线程空闲超过一定时间后，会被销毁)，其中临时线程最大数量=maximumPoolSize - corePoolSize，空闲最大时间由keepAliveTime控制，如果设置allowCoreThreadTimeOut(true)，临时线程永不销毁；

4、当线程总数达到maximumPoolSize 时，后续提交的任务都会被RejectedExecutionHandler拒绝。

为什么强制要求使用ThreadPoolExecutor创建线程池

在Executor类中，提供了以下场景的创建线程池的方法：

1、newCachedThreadPool，特点：必要时创建新线程，空闲线程会保留60s；

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

根据上面的ThreadPoolExecutor，允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM，还无法指定拒绝策略。

2、newFixedThreadPool，特点：包含固定的线程数，空闲线程会被一直保留；

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


public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM，还无法指定拒绝策略。

3、newSingleThreadExecutor，特点：只有一个线程的池，顺序执行每一个提交的任务；

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }



public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
}

在项目中创建线程池

如果需要在项目中创建线程池，那么应该将它设置成单例模式。

示例，用枚举的方式创建单例：

// 线程工厂
public class UserThreadFactory implements ThreadFactory {

	// 线程组命名标识
	private final String namePrefix;
	// 线程编号
	private final AtomicInteger nextId = new AtomicInteger(1);

	// 定义线程组名称，在 jstack 问题排查时，非常有帮助
	UserThreadFactory(String whatFeaturOfGroup) {
		namePrefix = "From UserThreadFactory's " + whatFeaturOfGroup + "-Worker-";
	}

	@Override
	public Thread newThread(Runnable task) {
		String name = namePrefix + nextId.getAndIncrement();
		Thread thread = new Thread(task, name);
		return thread;
	}

}


// 单例线程池
public enum ThreadPoolEnum {

	INSTANCE;
	
	private ThreadPoolExecutor threadPoolExecutor;

	// 枚举的特性,在JVM中只会被实例化一次
	private ThreadPoolEnum() {
		threadPoolExecutor = new ThreadPoolExecutor(2, 
				4, 10, TimeUnit.SECONDS, new PriorityBlockingQueue<Runnable>(), 
				new UserThreadFactory("first-threadPool"),
				new ThreadPoolExecutor.DiscardPolicy());
	}
	
	public ThreadPoolExecutor getInstance() {
        return threadPoolExecutor;
    }
	
	public static void main(String[] args) throws InterruptedException {
		ThreadPoolExecutor executor1 = ThreadPoolEnum.INSTANCE.getInstance();
		ThreadPoolExecutor executor2 = ThreadPoolEnum.INSTANCE.getInstance();
		System.out.println(executor1 == executor2);
		Runnable r = () -> {
			System.out.println("新创建的线程任务,交由线程池执行");
			System.out.println(Thread.currentThread().getName());
		};
		executor1.execute(r);
		// 销毁线程池
		executor1.shutdown();
	}
	
}

执行结果：

在Spring中创建线程池

@Configuration
public class ThreadPoolConfig {

	// 线程池的参数配置可由外部配置引入
	@Bean(destroyMethod = "shutdown")
	public ThreadPoolExecutor initThreadPoolExecutor() {
		return new ThreadPoolExecutor(2, 
				4, 10, TimeUnit.SECONDS, new PriorityBlockingQueue<Runnable>(), 
				new ThreadPoolExecutor.DiscardPolicy());
	}
	
}

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

最新评论