任务的执行与线程池

文章目录

• Executor 框架
• • 线程池
  • • 为什么不允许使用 Executors 去创建
    • 实现 Runnable 接口和 Callable 接口的区别
    • 执行 execute()方法和 submit()方法的区别是什么呢？
    • 四种线程池（统一都继承了 ThreadPoolExecutor 类）
    • 线程池的核心参数
    • 任务提交到线程池的流转
    • JDK提供的四种已有拒绝策略
    • 七种阻塞队列的实现
    • • 同步移交之 SynchronousQueue 的效果
  • Executor 的生命周期
  • 延迟任务与周期任务
• 任务的取消与关闭
• • 中断
  • 示例：计时运行
  • 使用 Future 来实现取消
  • shutdownNow 的局限性
• JVM 的关闭
• • 关闭钩子
  • 避免使用终结器
• 线程池的使用
• • 线程饥饿死锁
  • • • 线程池的大小
      • 配置 ThreadPoolExecutor 
  • 线程工厂
  • • • • 在调用构造函数后，再定制 ThreadPoolExecutor
• QA

Executor 框架

虽然 Executor 是个简单的接口，但它却为灵活且强大的异步任务执行框架提供了基础，该框架能支持多种不同类型的任务执行策略。它提供了一种标准的方法将任务的提交过程与执行过程解耦开来，并用Runnable来表示任务。Executor的实现还提供了对生命周期的支持，以及统计信息收集、应用程序管理机制和性能监视等机制。
Executor 基于生产者-消费者模式，提交任务的操作相当于生产者(生成待完成的工作单
元)，执行任务的线程则相当于消费者(执行完这些工作单元)。如果要在程序中实现-一个生产者-消费者模型，那么最简单的方式就是使用 Executor

每当看到下面这种形式的代码的时候：
new Thread(Runnable).start()
希望有一种更灵活的执行策略时，考虑使用 Executor 来代替 Thread。

线程池

为什么不允许使用 Executors 去创建

Executors 返回线程池对象的弊端如下：

• FixedThreadPool 和 SingleThreadExecutor ： 允许请求的队列长度为 Integer.MAX_VALUE ，可能堆积大量的请求，从而导致 OOM。
• CachedThreadPool 和 ScheduledThreadPool ： 允许创建的线程数量为 Integer.MAX_VALUE ，可能会创建大量线程，从而导致 OOM。

实现 Runnable 接口和 Callable 接口的区别

Runnable 接口 不会返回结果或抛出检查异常，但是 Callable 接口 可以。所以，如果任务不需要返回结果或抛出异常推荐使用 Runnable 接口 ，这样代码看起来会更加简洁。

执行 execute()方法和 submit()方法的区别是什么呢？

• execute()方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功与否；
• submit()方法用于提交需要返回值的任务。线程池会返回一个 Future 类型的对象，通过这个 Future 对象可以判断任务是否执行成功，并且可以通过 Future 的 get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用 get(long timeout，TimeUnit unit)方法则会阻塞当前线程一段时间后立即返回，这时候有可能任务没有执行完。

四种线程池（统一都继承了 ThreadPoolExecutor 类）

• newFixedThreadPool 创建一个固定线程数量的线程池。该线程池中的线程数量始终不变，可控制线程最大并发数，超出的线程会在队列中等待，如果线程异常退出，那么会补充一个新的。（newFixedThreadPool工厂方法将线程池的基本大小和最大大小设置为参数中指定的值，而且创建的线程池不会超时。）
• newCachedThreadPool 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。（newCachedThreadPool 工厂方法将线程池的最大大小设置为Integer.MAX_ VALUE，而将基本大小设置为零，并将超时设置为1分钟，这种方法创建出来的线程池可以被无限扩展，并且当需求降低时会自动收缩。）
• newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
• newScheduledThreadPool 创建一个固定线程数量的线程池，支持定时及周期性任务执行，类似于 Timer

线程池的核心参数

1. corePoolSize：指定了线程池中的常驻线程数量。
2. maximumPoolSize：指定了线程池中的最大线程数量。
3. keepAliveTime：当前线程池数量超过 corePoolSize 时，多余的空闲线程的存活时间，即多
   长时间内会被销毁。
4. unit ：keepAliveTime 的单位。
5. workQueue：阻塞队列，被提交但尚未被执行的任务。
6. threadFactory：线程工厂，用于创建线程，一般用默认的即可（如果需要命名什么的，就自定义）。
7. handler：拒绝策略，当任务太多来不及处理，如何拒绝任务。

任务提交到线程池的流转

JDK提供的四种已有拒绝策略

七种阻塞队列的实现

主要其实分为：

（1）有界 （2）无界 （3）同步移交

同步移交之 SynchronousQueue 的效果

对于非常大的或者无界的线程他，可以通过使用 SynchronousQueue来避免任务排队，以及直接将任务从生产者移交给工作者线程。SynchronousQueue 不是一个真正的队列，而是一种在线程之间进行移交的机制。要将一个元素放入SynchronousQueue中，必须有另一个线程正在等待接受这个元素。

如果没有线程正在等待，并且线程池的当前大小小于最大值，那么ThreadPoolExecutor将创建一个新的线程，否则根据饱和策略，这个任务将被拒绝。使用直接移交将更高效，因为任务会直接移交给执行它的线程，而不是被首先放在队列中，然后由工作者线程从队列中提取该任务。只有当线程池是无界的或者可以拒绝任务时，SynchronousQueue才有实际价值。在 newCachedThreadPool 工厂方法中就使用了 SynchronousQueue 。

为什么是阻塞队列？而不是一般队列？

（1）阻塞提交任务线程 和 在队列为空时，阻塞消费线程。
（2）自带阻塞和唤醒的功能。

为什么任务被先提交到阻塞队列，然后才创建小于最大线程数目 的线程去执行？

• 因为创建新线程的时候，是需要获取全局锁的，其他的线程就会被暂停（批量的概念）。

Executor 的生命周期

JVM 只有在所有线程全部终止，才会退出。因此，如果无法正确地关闭 Executor,那么JVM将无法结束。

线程池中的状态无非有三种。（1）正在运行（2）放在队列中待运行 （3）已运行完毕。因此其实也就存在两种关闭模式，一种是不接受新的请求，执行完所有正在运行与待运行的线程，另一种是全部粗暴关闭。既然Executor是为应用程序提供服务的，因而它们也是可关闭的(无论采用平缓的方式还是粗暴的方式)，并将在关闭操作中受影响的任务的状态反馈给应用程序。
为了解决执行服务的生命周期问题，Executor 扩展了ExecutorService 接口，添加了一些用于生命周期管理的方法(同时还有- -些用于任务提交的便利方法)。在程序清单6-7中给出了ExecutorService中的生命周期管理方法。

延迟任务与周期任务

为什么不使用Timer类而使用 newScheduledThreadPool

1. Timer在执行所有定时任务时只会创建一个线程。 如果某个任务的执行时间过长，那么将破坏其他TimerTask的定时精确性。例如某个周期TimerTask需要每l0ms执行一次，而另一个TimerTask需要执行40ms，那么这个周期任务或者在40ms任务执行完成后快速连续地调用4.次，或者彻底“丢失”4次调用(取决于它是基于固定速率来调度还是基于固定延时来调度)。.线程池能弥补这个缺陷，它可以提供多个线程来执行延时任务和周期任务。
2. Timer的另一个问题是，如果TimerTask抛出了一个未检查的异常，那么Timer将表现出糟糕的行为。Timer线程并不捕获异常，因此当TimerTask抛出未检查的异常时将终止定时线程。这种情况下，Timer也不会恢复线程的执行，而是会错误地认为整个Timer都被取消了。因此，已经被调度但尚未执行的TimerTask将不会再执行，新的任务也不能被调度。(这个问题称之为“线程泄漏[Thread Leakage]"，7.3 节将介绍该问题以及如何避免它。)

任务的取消与关闭

中断

在程序清单中的BrokenPrimeProducer就说明了这个问题。生产者线程生成素数，并将它们放入一个阻塞队列。如果生产者的速度超过了消费者的处理速度，队列将被填满，put方法也会阻塞。当生产者在put方法中阻塞时，如果消费者希望取消生产者任务，那么将发生什么情况?它可以调用cancel方法来设置cancelled标志，但此时生产者却永远不能检查这个标志，因为它无法从阻塞的put方法中恢复过来(因为消费者此时已经停止从队列中取出素数,所以put方法将一直保持阻塞状态)。

// 生产者
class BrokenPrimeProducer extends Thread {
   
    private final BlockingQueue<BigInteger> queue;
    private volatile boolean cancelled = false;

    BrokenPrimeProducer(BlockingQueue<BigInteger> queue) {
   
        this.queue = queue;
    }

    public void run() {
   
        try {
   
            BigInteger p = BigInteger.ONE;
            while (!cancelled)
                // 生产素数
                queue.put(p = p.nextProbablePrime());
        } catch (InterruptedException consumed) {
   
        }
    }

    public void cancel() {
   
        cancelled = true;
    }
}

通常，中断是实现取消最合理的方式。
除非拥有某个线程，否则不能对该线程进行操控。

示例：计时运行

使用 Future 来实现取消

shutdownNow 的局限性

无法通过常规方法来找出哪些任务已经开始但尚未结束。这意味着我们无法在关闭过程中知道正在执行的任务的状态，除非任务本身会执行某种检查。要知道哪些任务还没有完成，你不仅需要知道哪些任务还没有开始，而且还需要道当Executor关闭时哪些任务正在执行。

JVM 的关闭

正常关闭：system.exit()，最后一个守护线程结束，ctrl-c
异常关闭：kill -9

关闭钩子

• 在正常关闭中，JVM首先调用所有已注册的关闭钩子(Shutdown Hook)。关闭钩子是指通过 Runtime.addShutdownHook 注册的但尚未开始的线程。JVM并不能保证关闭钩子的调用顺序。

• 在关闭应用程序线程时，如果有(守护或非守护)线程仍然在运行，那么这些线程接下来将与关闭进程并发执行。当所有的关闭钩子都执行结束时，如果 runFinalizersOnExit 为true,那么JVM 将运行终结器，然后再停止。JVM并不会停止或中断任何在关闭时仍然运行的应用程序线程。

• 当JVM最终结束时，这些线程将被强行结束。如果关闭钩子或终结器没有执行完成，那么正常关闭进程“挂起”并且JVM必须被强行关闭。当被强行关闭时，只是关闭JVM，而不会运行关闭钩子。

• 关闭钩子应该是线程安全的:它们在访问共享数据时必须使用同步机制，并且小心地避免发生死锁，这与其他并发代码的要求相同。而且，关闭钩子不应该对应用程序的状态(例如，其他服务是否已经关闭，或者所有的正常线程是否已经执行完成)或者JVM的关闭原因做出任何假设，因此在编写关闭钩子的代码时必须考虑周全。最后，关闭钩子必须尽快退出，因为它们会延迟JVM的结束时间，而用户可能希望JVM能尽快终止。.

• 关闭钩子可以用于实现服务或应用程序的清理工作，例如删除临时文件，或者清除无法由操作系统自动清除的资源。在程序清单7-26中给出了如何使程序清单7-16中的LogService在其start方法中注册-一个关闭钩子，从而确保在退出时关闭日志文件。由于关闭钩子将并发执行，因此在关闭日志文件时可能导致其他需要日志服务的关闭钩子出现问题，解决这个问题的办法是：将针对于同一资源的关闭钩子，统一写在一个关闭钩子中，从而也避免了并发导致的一系列问题。
  
  

避免使用终结器

线程池的使用

线程饥饿死锁

1. 在一个线程池中，如果一个任务依赖于其他任务的执行，就可能产生死锁。对应一个单线程话的Executor,一个任务将另一个任务提交到相同的Executor中，并等待新提交的任务的结果，这总会引发死锁。第二个任务滞留在工作队列中，直到第一个任务完成，但是第一个任务不会完成，因为它在等待第二个任务的完成。
2. 同样在一个大的线程池中，如果所有线程执行的任务都阻塞在线程池中，等待着仍然处于同一个工作队列中的其他任务，那么会发生同样的问题。这就是线程饥饿死锁。

public class ThreadDeadlock {
   
	  ExecutorService exec = Executors.newSingleThreadScheduledExecutor();
//	  ExecutorService exec = Executors.newCachedThreadPool();    //如果添加给线程池中添加足够多的线程，就可以让所有任务都执行，避免饥饿死锁。
	 /**
	  * 模拟页面加载的例子
	  * 
	  * 产生死锁分析：
	  * RenderPageTask任务中有2个子任务分别是“加载页眉”和“加载页脚”。当提交RenderPageTask任务时，实际上是向线程池中添加了3个任务，
	  * 但是由于线程池是单一线程池，同时只会执行一个任务，2个子任务就会在阻塞在线程池中。而RenderPageTask任务由于得不到返回，也会
	  * 一直堵塞，不会释放线程资源让子线程执行。这样就导致了线程饥饿死锁。
	  * 
	  * 在一个Callable任务中，要返回2个子任务
	  * @author hadoop
	  *
	  */
	  class RenderPageTask implements Callable<String>{
   
		@Override
		public String call() throws Exception {
   
			Future<String>  header,footer;
			header = exec.submit(new Callable<String>(){
   
				@Override
				public String call() throws Exception {
   
					System.out.println("加载页眉");
					Thread.sleep(2*1000);
					return "页眉";
				}
			});
			footer = exec.submit(new Callable<String>(){
   
				@Override
				public String call() throws Exception {
   
					System.out.println("加载页脚");
					Thread.sleep(3*1000);
					return "页脚";
				}
			});
			System.out.println("渲染页面主体");
			return header.get() + footer.get();
		} 
	  }
	public static void main(String[] args) throws InterruptedException, ExecutionException {
   
		ThreadDeadlock td = new ThreadDeadlock();
		Future<String>  futre = td.exec.submit(td.new RenderPageTask());
		String result = futre.get();
		System.out.println("执行结果为："  +  result);
		
	}
}

线程池的大小

如何获得 CPU 数目：System.out.println(Runtime.getRuntime().availableProcessors());

配置 ThreadPoolExecutor

• 基本大小：工作队列满了，才会开始创建超出这个数量的线程，最大到 最大大小
• 最大大小：如果线程池的数量超过了最大大小，且有线程是能够被回收的，那么就线程就会被回收
• 工作队列：（1）有界 （2）无界 （3）同步移交
• 饱和策略：当有界队列被填满后，饱和策略开始发挥作用。ThreadPoolExecutor 的饱和策略可以通过调用setRejectedExecutionHandler 来修改。( 如果某个任务被提交到一个已被关闭的Executor 时，也会用到饱和策略。) JDK提供了几种不同的 RejectedExecutionHandler 实现，每种实现都包含有不同的饱和策略: AbortPolicy、 CallerRunsPolicy、 DiscardPolicy 和 DiscardOldestPolicy。

线程工厂

线程池需要创建一个线程时，都是从线程工厂来获得一个线程的。因为可以设置 setUncaughtExceptionHandler，天然带有调试信息，取一个名称或者存储错误日志。

如果在应用程序中需要利用安全策略来控制对某些特殊代码库的访问权限，那么可以通过Executor中的privilegedThreadFactory 工厂来定制自己的线程工厂。通过这种方式创建出来的线程，将与创建privilegedThreadF actory的线程拥有相同的访问权限、AccessControlContext 和 contextClassLoader。如果不使用 privilegedThreadFactory,线程池创建的线程将从在需要新线程时调用execute或submit的客户程序中继承访问权限，从而导致令人困惑的安全性异常。

在调用构造函数后，再定制 ThreadPoolExecutor

QA

• 创建过多的线程会导致什么问题？

1. 线程的创建，销毁，生命周期长
2. 资源消耗。大量线程闲置会导致内存，GC，等其他性能消耗
3. 稳定性。系统限制，系统稳定性不足。

• Timer 的缺陷

1. Timer支持基于绝对时间而不是相对时间的调度机制，因此任务的执行对系统时钟变化很敏感，而
   ScheduledThreadPoolExecutor只支持基于相对时间的调度。

https://tech.meituan.com/2020/04/02/java-pooling-pratice-in-meituan.html