文章目录

• 概述
• 概述
• Executor与线程池
• Java 中的线程池
• 使用线程池的注意事项
• • 强烈建议使用有界队列
  • 默认拒绝策略要慎重使用
  • 注意异常处理的问题
• 如何获取任务执行结果
• • Future接口的5个方法
  • ThreadPoolExecutor的3种submit方式
  • FutureTask 工具类
  • • 使用 FutureTask
    • • FutureTask 对象提交给 ThreadPoolExecutor 去执行
      • FutureTask 对象直接被 Thread 执行
• FutureTask实战---烧水泡茶

概述

最近在阅读耗子叔的《左耳听风》 ， 记一些小笔记

概述

在 Java 中，在 JDK 1.8 里也引入了类似 JavaScript 的玩法 —— CompletableFuture。这个类提供了大量的异步编程中 Promise 的各种方式。

CompletableFuture.supplyAsync(this::findReceiver).thenApply(this::sendMsg).thenAccept(this::notify);

• supplyAsync() 表示执行一个异步方法
• thenApply() 表示执行成功后再串联另外一个异步方法
• 最后是 thenAccept() 来处理最终结果。

接下来，我们再来看一下，Java 这个类相关的异常处理：

CompletableFuture.supplyAsync(Integer::parseInt) //输入: "ILLEGAL".thenApply(r -> r * 2 * Math.PI).thenApply(s -> "apply>> " + s).exceptionally(ex -> "Error: " + ex.getMessage());

注意到上面代码里的 exceptionally() 方法 , 运行上面的代码，会出现如下输出：

Error: java.lang.NumberFormatException: For input string: "ILLEGAL"

也可以这样：

CompletableFuture.supplyAsync(Integer::parseInt) // 输入: "ILLEGAL".thenApply(r -> r * 2 * Math.PI).thenApply(s -> "apply>> " + s).handle((result, ex) -> {if (result != null) {return result;} else {return "Error handling: " + ex.getMessage();}});

上面代码中，使用了 handle() 方法来处理最终的结果，其中包含了异步函数中的错误处理。

Executor与线程池

创建对象，仅仅是在 JVM 的堆里分配一块内存而已；而创建一个线程，却需要调用操作系统内核的 API，然后操作系统要为线程分配一系列的资源，这个成本就很高了，所以线程是一个重量级的对象，应该避免频繁创建和销毁

为什么线程池没有采用一般意义上池化资源的设计方法呢？如果线程池采用一般意义上池化资源的设计方法，应该是下面示例代码这样。

//采用一般意义上池化资源的设计方法
class ThreadPool{// 获取空闲线程Thread acquire() {}// 释放线程void release(Thread t){}
} 
//期望的使用
ThreadPool pool；
Thread T1=pool.acquire();
//传入Runnable对象
T1.execute(()->{//具体业务逻辑......
});

可以来思考一下，假设我们获取到一个空闲线程 T1，然后该如何使用 T1 呢？

我们期望的可能是这样：通过调用 T1 的 execute() 方法，传入一个 Runnable 对象来执行具体业务逻辑，就像通过构造函数 Thread(Runnable target) 创建线程一样。可惜的是，翻遍 Thread 对象的所有方法，都不存在类似 execute(Runnable target) 这样的公共方法。

线程池是一种生产者 - 消费者模式。 线程池的使用方是生产者，线程池本身是消费者。在下面的示例代码中，创建了一个非常简单的线程池 MyThreadPool，可以通过它来理解线程池的工作原理。

//简化的线程池，仅用来说明工作原理
class MyThreadPool{//利用阻塞队列实现生产者-消费者模式BlockingQueue workQueue;//保存内部工作线程List threads   = new ArrayList<>();// 构造方法MyThreadPool(int poolSize,  BlockingQueue workQueue){this.workQueue = workQueue;// 创建工作线程for(int idx=0; idxWorkerThread work = new WorkerThread();work.start();threads.add(work);}}// 提交任务void execute(Runnable command){workQueue.put(command);}// 工作线程负责消费任务，并执行任务class WorkerThread extends Thread{public void run() {//循环取任务并执行while(true){ // ①Runnable task = workQueue.take();task.run();} }}  
}/** 下面是使用示例 **/
// 创建有界阻塞队列
BlockingQueue workQueue =   new LinkedBlockingQueue<>(2);
// 创建线程池  
MyThreadPool pool = new MyThreadPool(  10, workQueue);
// 提交任务  
pool.execute(()->{System.out.println("hello");
});

在 MyThreadPool 的内部，我们维护了一个阻塞队列 workQueue 和一组工作线程，工作线程的个数由构造函数中的 poolSize 来指定。用户通过调用 execute() 方法来提交 Runnable 任务，execute() 方法的内部实现仅仅是将任务加入到 workQueue 中。MyThreadPool 内部维护的工作线程会消费 workQueue 中的任务并执行任务，相关的代码就是代码①处的 while 循环。线程池主要的工作原理就这些.

Java 中的线程池

Java 提供的线程池相关的工具类中，最核心的是 ThreadPoolExecutor，通过名字你也能看出来，它强调的是 Executor，而不是一般意义上的池化资源。

最完备的构造函数有 7 个参数

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

可以把线程池类比为一个项目组，而线程就是项目组的成员

• corePoolSize：表示线程池保有的最小线程数。有些项目很闲，但是也不能把人都撤了，至少要留 corePoolSize 个人坚守阵地。
• maximumPoolSize：表示线程池创建的最大线程数。当项目很忙时，就需要加人，但是也不能无限制地加，最多就加到 maximumPoolSize 个人。当项目闲下来时，就要撤人了，最多能撤到 corePoolSize 个人。
• keepAliveTime & unit：上面提到项目根据忙闲来增减人员，那在编程世界里，如何定义忙和闲呢？很简单，一个线程如果在一段时间内，都没有执行任务，说明很闲，keepAliveTime 和 unit 就是用来定义这个“一段时间”的参数。也就是说，如果一个线程空闲了keepAliveTime & unit这么久，而且线程池的线程数大于 corePoolSize ，那么这个空闲的线程就要被回收了
• workQueue：工作队列，和上面示例代码的工作队列同义。
• threadFactory：通过这个参数可以自定义如何创建线程，
• handler：通过这个参数你可以自定义任务的拒绝策略。如果线程池中所有的线程都在忙碌，并且工作队列也满了（前提是工作队列是有界队列），那么此时提交任务，线程池就会拒绝接收。至于拒绝的策略，可以通过 handler 这个参数来指定。

ThreadPoolExecutor 已经提供了以下 4 种策略。

CallerRunsPolicy：提交任务的线程自己去执行该任务。

AbortPolicy：默认的拒绝策略，会 throws RejectedExecutionException。

DiscardPolicy：直接丢弃任务，没有任何异常抛出。

DiscardOldestPolicy：丢弃最老的任务，其实就是把最早进入工作队列的任务丢弃，然后把新任务加入到工作队列。

Java 在 1.6 版本还增加了 allowCoreThreadTimeOut(boolean value) 方法，它可以让所有线程都支持超时，这意味着如果项目很闲，就会将项目组的成员都撤走。

使用线程池的注意事项

强烈建议使用有界队列

考虑到 ThreadPoolExecutor 的构造函数实在是有些复杂，所以 Java 并发包里提供了一个线程池的静态工厂类 Executors，利用 Executors 你可以快速创建线程池。不过目前大厂的编码规范中基本上都不建议使用 Executors 了。

不建议使用 Executors 的最重要的原因是：Executors 提供的很多方法默认使用的都是无界的 LinkedBlockingQueue，高负载情境下，无界队列很容易导致 OOM，而 OOM 会导致所有请求都无法处理，这是致命问题。所以强烈建议使用有界队列。

默认拒绝策略要慎重使用

使用有界队列，当任务过多时，线程池会触发执行拒绝策略，线程池默认的拒绝策略会 throw RejectedExecutionException 这是个运行时异常，对于运行时异常编译器并不强制 catch 它，所以开发人员很容易忽略。因此默认拒绝策略要慎重使用。如果线程池处理的任务非常重要，建议自定义自己的拒绝策略；并且在实际工作中，自定义的拒绝策略往往和降级策略配合使用。

注意异常处理的问题

使用线程池，还要注意异常处理的问题，例如通过 ThreadPoolExecutor 对象的 execute() 方法提交任务时，如果任务在执行的过程中出现运行时异常，会导致执行任务的线程终止；不过，最致命的是任务虽然异常了，但是你却获取不到任何通知，这会让你误以为任务都执行得很正常。虽然线程池提供了很多用于异常处理的方法，但是最稳妥和简单的方案还是捕获所有异常并按需处理，你可以参考下面的示例代码。

try {//业务逻辑
} catch (RuntimeException x) {//按需处理
} catch (Throwable x) {//按需处理
} 

如何获取任务执行结果

ThreadPoolExecutor 的 void execute(Runnable command) 方法，利用这个方法虽然可以提交任务，但是却没有办法获取任务的执行结果（execute() 方法没有返回值）。而很多场景下，我们又都是需要获取任务的执行结果的。

Java 通过 ThreadPoolExecutor 提供的 3 个 submit() 方法和 1 个 FutureTask 工具类来支持获得任务执行结果的需求。

// 提交Runnable任务
Future submit(Runnable task);
// 提交Callable任务
 Future submit(Callable task);
// 提交Runnable任务及结果引用  
 Future submit(Runnable task, T result);

它们的返回值都是 Future 接口.

Future接口的5个方法

Future 接口有 5 个方法

// 取消任务
boolean cancel(boolean mayInterruptIfRunning);
// 判断任务是否已取消  
boolean isCancelled();
// 判断任务是否已结束
boolean isDone();
// 获得任务执行结果
get();
// 获得任务执行结果，支持超时
get(long timeout, TimeUnit unit);

• 取消任务的方法 cancel()
• 判断任务是否已取消的方法 isCancelled()
• 判断任务是否已结束的方法 isDone()
• 2 个获得任务执行结果的 get() 和 get(timeout, unit)，其中最后一个 get(timeout, unit) 支持超时机制。

ThreadPoolExecutor的3种submit方式

这 3 个 submit() 方法之间的区别在于方法参数不同，

• 提交 Runnable 任务 submit(Runnable task) ：这个方法的参数是一个 Runnable 接口，Runnable 接口的 run() 方法是没有返回值的，所以 submit(Runnable task) 这个方法返回的 Future 仅可以用来断言任务已经结束了，类似于 Thread.join()。

• 提交 Callable 任务 submit(Callable task)：这个方法的参数是一个 Callable 接口，它只有一个 call() 方法，并且这个方法是有返回值的，所以这个方法返回的 Future 对象可以通过调用其 get() 方法来获取任务的执行结果。

• 提交 Runnable 任务及结果引用 submit(Runnable task, T result)：这个方法很有意思，假设这个方法返回的 Future 对象是 f，f.get() 的返回值就是传给 submit() 方法的参数 result。

  这个方法该怎么用呢？下面这段示例代码展示了它的经典用法。需要注意的是 Runnable 接口的实现类 Task 声明了一个有参构造函数 Task(Result r) ，创建 Task 对象的时候传入了 result 对象，这样就能在类 Task 的 run() 方法中对 result 进行各种操作了。result 相当于主线程和子线程之间的桥梁，通过它主子线程可以共享数据

ExecutorService executor  = Executors.newFixedThreadPool(1);
// 创建Result对象r
Result r = new Result();
r.setAAA(a);
// 提交任务
Future future = executor.submit(new Task(r), r);  
Result fr = future.get();
// 下面等式成立
fr === r;
fr.getAAA() === a;
fr.getXXX() === xclass Task implements Runnable{Result r;//通过构造函数传入resultTask(Result r){this.r = r;}void run() {//可以操作resulta = r.getAAA();r.setXXX(x);}
}

FutureTask 工具类

Future是一个接口，FutureTask是一个工具类。

FutureTask的构造函数

FutureTask(Callable callable);
FutureTask(Runnable runnable, V result);

FutureTask实现了Runnable（可以将FutureTask对象作为任务提交给ThreadPoolExecutor执行 & 直接被Thread执行）和Future接口（获得任务的执行结果）。 利用FutureTask对象可以很容易获取子线程的执行结果。

使用 FutureTask

其实很简单，FutureTask 实现了 Runnable 和 Future 接口

• 由于实现了 Runnable 接口，所以可以将 FutureTask 对象作为任务提交给 ThreadPoolExecutor 去执行，也可以直接被 Thread 执行；
• 又因为实现了 Future 接口，所以也能用来获得任务的执行结果

FutureTask 对象提交给 ThreadPoolExecutor 去执行

public static void submitFutureTask() throws ExecutionException, InterruptedException{
// 创建FutureTask
FutureTask futureTask = new FutureTask<>(()->6*6);

    // 创建线程池ExecutorService es = Executors.newFixedThreadPool(1);// 提交FutureTaskes.submit(futureTask);// 获取计算结果Integer integer = futureTask.get();System.out.println(integer);// 关闭线程池es.shutdown();}

FutureTask 对象直接被 Thread 执行

    public static void executeFutureTaskByThread() throws ExecutionException, InterruptedException {// 创建FutureTaskFutureTask fs = new FutureTask<>(()->6*8);// 创建线程Thread thread = new Thread(fs,"TestThread");// 提交thread.start();// 获取返回结果Integer integer = fs.get();System.out.println(integer);}

利用 FutureTask 对象可以很容易获取子线程的执行结果

FutureTask实战—烧水泡茶