Java异步编程实战
认识异步编程
异步编程概念与作用
-
同步编程简单，但是容易出现性能瓶颈；并行多个线程可以提高系统性能，但对共享资源进行访问时引入资源竞争和并发问题
- 使用同步阻塞的编程方式会浪费资源
- 通过增加单机系统的线程个数的并行编程方式并不是最好解决方式。异步、非阻塞的代码可以进一步提高系统性能。
-
异步编程是可以让程序并行玉兴的一种手段，其可以让程序中的一个工作但与与主应用程序线程分开独立余小宁，并且在工作单元运行结束后，会通知主应用程序线程他的运行结果或者失败原因。
- 使用异步编程可以提高应用程序的性能和响应能力等
异步编程场景
-
需要异步处理一些事情，而不需要知道异步任务的结果
- 例如在调用栈里异步打印日志，为了不让日志打印阻塞调用线程，会把日志设置为异步方式（基于一个内存队列实现的多生产单消费模型）
-
使用
Java中可以直接开启一个线程来实现，也可以把异步任务封装为任务对象投递到线程池中来执行
在Spring框架中提供了@Async注解把一个任务异步化来进行处理
Future可以获取异步执行的结果，但是会阻塞调用线程，并没有实现完全异步化
JDK8中提供了CompletableFuture弥补Future的缺点。允许以非阻塞方式和基于通知的方式处理结果，其通过设置回调函数方式，让主线程彻底解放出来，实现了实现了意义上的异步处理。
-
JDK8 引入Stream，有效的处理数据流，其使用声明方式编程让我们可以写出可读性、可维护性很强的代码，并且结合CompletableFuture完美的实现异步编程
- 缺点是它产生的流只能使用一次，并且缺少与时间相关的操作（例如RxJava中基于时间窗口的缓存元素），虽然可以执行计算，但无法指定要使用的线程池。同时它也没有设计用于处理延迟的操作（例如RxJava中的defer操作）
- Reactor、RxJava等Reactive API就是为了解决这些问题而生的
Reactor、RxJava等反应式API也提供了Java 8 Stream的运算符，但他们更适用于流序列，并允许定义一个转换操作的管道，该管道应用于通过它的数据。Reactive旨在处理同步或异步操作，并允许你对元素进行缓冲（buffer）、合并（merge）、连接（join）等各种转换。
-
上面都是单JVM内的异步编程，跨网络的交互式
- 例如线程A顺序的对多个服务请求进行调用，简单但是浪费资源。好的做法应该是在发起请求的调用线程发起请求后，注册一个回调函数，然后马上返回去执行其他操作，当远端把结果返回后再使用IO线程或框架线程池中的线程执行回调函数。
-
Java中NIO可以实现网络中的异步调用，高性能异步、基于事件驱动的网络编程框架Netty的出现从繁杂的Java NIO程序中解放出来
- Netty框架将网络编程逻辑与业务逻辑处理分离开来，在内部帮我们自动处理好网络与异步处理逻辑，让我们转移写自己的业务处理罗，而Netty得异步非阻塞能力与CompletableFuture结合则可以轻松的实现网络请求的异步调用。
- 异步调用两个服务，然后借助CompletabelFuture的能力等两次RPC调用都异步返回结果后再执行其他操作。
有了CompetableFuture实现异步编程，可以很自然的使用适配器来实现Reactive风格的编程。当我们使用RxJava API时，只需要使用Flowable的一些函数转换为CompletableFuture为Flowable对象即可。
Servlet 3.0规范让Servlet的执行变为了异步，但是其IO还是阻塞式的
Servlet 3.1规范提供了非阻塞IO来解决这个问题
-
Servlet实现了异步处理与非阻塞IO，但其异步是步彻底的
- 规范是同步的（Filter，Servlet）或阻塞的（getParameter，getPart)
-
WebFlux是与Servlet技术栈并存的一种新技术
- 基于JDK8函数时编程，与Netty实现天然的异步、非阻塞处理
- 使用少量线程和较少的硬件资源来处理并发的非阻塞Web技术栈
-
降低异步编程成本的框架
-
disruptor
- 高性能线程间消息传递库Disruptor，其通过事件预先分配内存、无锁CAS算法、缓冲行填充、两阶段协议提交来实现多线程并发处理不同的元素，从而实现高性能的异步处理
-
Akka
- 基于Actor模式实现了天然支持分布式的使用消息进行异步处理的服务
-
Apache RocketMetaQ
- 实现了应用间的异步解耦、流量削峰
-
-
Go语言
- 语言层面内置的goroutine与channel可以轻松实现复杂的异步处理能力
总结
显示使用线程何线程池实现异步编程
显示使用线程实现异步编程
同步编程模型例子
-
使用Thread创建一个线程来异步执行任务
-
开启线程的方法
- 实现Iainjava.lang.Runnable接口的run方法，然后传递Runnable接口的实现类作为创建Thread时的参数，启动线程
- 实现Thread类，并重写run方法
-
存在的问题
-
生成环境不建议直接创建Thread，因为线程创建与销毁是有开销的，并且没有限制下称的个数
- 线程池中的线程可以被复用，可以大大减少线程创建与销毁开销，可以有效限制创建的县城个数
-
Thread执行的任务没有返回值
- Future解决
-
增加编程负担
- Java提供了很多分装的类库
-
-
实现Thread类，并重写run方法
显示使用线程池实现异步编程
-
如何显示使用线程池实现异步编程
-
把任务放到线程池里异步执行
- CallerRunsPolicy当线程池任务包和，执行拒绝策略时不会丢弃新的任务，而是会使用调用线程来执行
- 使用了命名的线程创建工厂，以便排查问题时可以方便追溯是哪个相关业务
-
-
线程池ThreadPoolExecutro原理剖析
-
概述
-
类图结构
- 子主题 1
-
ThreadPoolExecutor解析
-
成员变量ctl是Integer的原字变量，使用一个变量同时记录线程池状态和线程池中线程个数
- 高3位表示线程池状态，后面29位用来记录线程池线程个数
-
线程池主要状态列举
-
RUNNING
- 接收新任务并且处理阻塞队列里的任务
-
SHUTDOWN
- 拒绝新任务但是处理阻塞队列里的任务
-
STOP
- 拒绝新任务并且抛弃阻塞队列里的任务，同时中断正在处理的任务
-
TIDYING
- 所有任务都执行完（包含阻塞队列里面的任务），当前线程池活动线程位0， 将要调用terminated方法
-
TERMINATED
- 终止状态。terminated方法调用完成以后的状态
-
-
线程池状态之间转换路径
-
RUNNING->SHUTDOWN
- 当显示调用shutdown()方法时，或者因式调用了finalize()，它里面调用了shutdown方法时
-
RUNNING或者SHUTDOWN->STOP
- 当显示调用shutdownNow()方法时
-
SHUTDOWN->TIDYING
- 当线程池和任务队列都为空时
-
STOP->TERMINATED
- 当terminated()hook方法执行完成时
-
-
获取线程池的运行状态和线程池中线程个数
- runStateOf(int c)
- workerCountOf(int c)
- ctlOf(int rs, int wc)
-
线程池的配置参数
-
corePoolSize
- 线程池核心线程个数
-
workQueue
-
用于保存等待执行的任务的阻塞队列
- ArrayBlockQueue
- LinkedBlockingQueue
- SynchronousQueue
- PriorityBlockingQueue
- DelayQueue
- LinkedBlockingDeque
- LinkedTransferQueue
-
-
maximunPoolSize
- 线程池最大线程数量
-
threadFactory
- 创建线程的工厂类
-
defaultHandler
-
饱和策略（拒绝策略），当队列满了并且线程个数达到最大maximunPoolSize后采取的策略
- AbortPolicy（抛出异常）
- CallerRunsPolicy（使用调用者所在线程来运行任务）
- DiscardOldestPolicy（调用poll丢弃一个任务，执行当前任务）
- DiscardPolicy（默默丢弃，不抛出异常）
-
-
keepAliveTime
- 存货事件。如果当前线程池中的线程比核心线程数量要多，并且是闲置状态的化，这些限制的线程能存活的最大事件
-
-
mainLock
- ReentrantLock，用来控制新增Worker线程时的原子性，termination是锁对应的条件队列，在线程调用awaitTermination时用来存放阻塞的线程
-
-
Worker解析
-
Worker继承AQS和Runnable接口，时具体承载任务的对象。
- Worker继承了AQS，实现了简单不可重入独占锁，其中state=0标示锁未被获取的状态，state=1标示锁已经被获取的状态，state=-1时创建Worker时默认的状态。创建时状态设置为-1是为了避免该线程在运行runWorker()方法前被中断
firstTask记录该工作线程执行的第一个任务
Thread是具体执行任务的线程
-
-
DefaultThreadFactory
- 线程哦工厂，newThread方法是对线程的一个修饰
- poolNumber是一个静态的原字变量，用来统计线程工厂的个数
- threadNumber是用来记录每个线程工厂创建了多少线程
- 这两个值也作为线程池和线程的名称的一部分
-
ThreadPoolExecutro提供了一系列构造函数让我们创建线程池
- 子主题 1
-
-
提交任务到线程池原理解析
-
ThreadPoolExecutor中提交任务到线程池的方法
- void execute(Runnable)
- Future<?> submit(Runnable)
- Future submit(Runnable Task, T result)
- Future submit(Callable)
-
void execute(Runnable)提交任务到线程池的逻辑
- 代码
- 流程图
-
Future<? submit(Runnable)提交任务的逻辑
- 代码
- 流程图
-
Future sumit(Runnable Task, T result)逻辑
- 代码
- 流程图
-
-
线程池中任务执行原理解析
当用户线程提交任务到线程池后，在线程池没有执行拒绝策略的情况下，用户线程会马上返回，而提交的任务要么直接切换到线程池中的Worker线程来执行，要么先放入线程池的阻塞队里面
-
Worker线程工厂原理
- 构造函数
-
关闭线程池原理解析
-
线程池中有两种模式的线程池关闭方法
-
shutdown()
- void
-
shutdownnow()
- 返回值为这时队列里面被丢弃的任务列表
-
shutdown()代码逻辑
shutdownNow()代码逻辑
-
-
线程池的拒绝策略解析
当线程池处于包和状态时，再向线程池投递任务，而对于投递的任务如何处理，是由线程池拒绝策略决定的。拒绝策略的执行是execute方法
-
RejectedExecutionHandler接口
- 用来提供对线程池拒绝策略的抽象
-
分类
-
AbortPolicy
- 源码
-
CallerRunsPolicy
- 源码
-
DiscardPolicy
- 源码
-
DiscardOldestPolicy
- 源码
-
-
总结
- 讨论Java中最基础的显示创建线程的方式来实现异步编程，并指出了其存在的三个问题；显示使用线程池来实现异步编程，并且讲解了线程池的实现原理；目前获取值时还是要阻塞调用线程池的，可以用CompletableFuture来解决
基于JDK中的Future实现异步编程
JDK中的Future
-
介绍
- JUC中Future代表着异步计算结果，Future提供了一系列方法用来检查计算结果是否已经完成，提供了同步等待任务执行完成的方法，提供了获取计算结果的方法。
-
类图
- 子主题 1
-
5个接口方法
-
V get()
- 阻塞式等待异步计算任务完成，并返回结果；如果等待过程中有其他线程取消了该任务，则调用线程抛出CancellationException异常；如果再等待结果的过程中有其他线程中断了该线程，则调用线程抛出InterruptedException异常；如果任务计算过程中抛出了异常，则调用线程会抛出ExecutionException异常
-
V get(long timeout, TimeUnit unit)
- 与get()的区别式，调用线程不会一直被阻塞。添加超时能避免调用线程死等地归情况，让调用线程可以及时释放
-
boolean isDone()
- 如果计算任务已经完成则返回true，否则返回false。这里任务完成是指正常完成、由于抛出异常而完成了或者任务被取消了
-
boolean cancel(boolean)
- 尝试取消任务的执行；如果当前任务已经完成或者任务已经被取消了，则尝试取消任务会失败；如果任务还没被执行时调用了取消任务，则任务将永远不会被执行；如果任务已经开始运行了，这是取消任务，则参数将决定是否要将正在执行任务的线程中断，如果为true则要中断，否则标示不中断；当调用取消任务后，再嗲用isDone()方法，后者返回true，随后调用isCancelled()方法也会一直返回true；如果任务不能被取消，比如任务完成后已经被取消了，则该方法返回false
-
boolean isCancelled()
- 如果任务再执行完毕前被取消了，则该方法返回true，否则返回false
-
-
使用建议
- 超时等待配合取消使用
JDK中的FutureTask
-
FutureTask概述
- FutureTask代表了一个可悲取消的异步计算任务，该类实现Future接口
- FutureTask任务的结果只有当任务完成后才能获取，只能通过get系列方法获取，当结果还没出来时，线程调用get系列方法不i被阻塞。一旦任务被执行完成，任务将不能成功其，除非运行时使用了runAndReset方法。FutureTask的任务类型可以时Callable类型或者Runnable类型。FutureTask类型的任务可以被提交到线程池执行
-
FutureTask执行方式
作为Runnable传入Thread中
-
放入线程池
execute()
-
submit()
- 更简单，但是返回的是Future，不是FutureTask
-
FutureTask的类图结构
-
类图
- 子主题 1
-
概要说明
- FutureTask实现Future接口和Runnable接口，所以可以执行任务，可以传到线程池胡总和线程中执行
-
详细说明
-
state变量
是使用volatile关键字修饰的int变量（用来解决多线程下内存不可见问题），用来记录任务状态
-
任务状态枚举
- NEW
- COMPLETING
- NORMAL
- EXCEPTIONAL
- CANCELLED
- INTERRUPTING
- INTERRUPTED
-
状态转换说明
- 一开始任务状态会被初始化为NEW；当通过set、setException、cancel函数设置任务结果时，任务会转换为终止状态；再任务完成过程中，任务状态可能会变为COMPLETING(当结果被使用set方法设置时），也可能会经过INTERRUPTING状态(当使用cancel(true)方法取消任务并中断任务时）。当任务被中断后，任务状态为INTERRUPTED;当任务被取消后，任务状态为CANCELLED；当任务正常终止时，任务状态为NORMAL；当任务执行异常后，任务状态会变为EXCEPTIONAL。
-
在任务运行过程中，任务可能的状态转换路径
-
NEW->COMPETING->NORMAL
- 正常终止流程转换
-
NEW-COMPLETING->EXCEPTIONAL
- 执行过程中发生异常流程转换
-
NEW->CANCELLED
- 任务还没开始就被取消
-
NEW->INTERRUPTING->INTERRUPTED
- 任务被中断
-
-
任务中的终止状态
- NORMAL\EXCEPTIONAL\CANCELLED\INTERRUPTED
callable是由返回值的可执行任务，创建FutureTask对象时，可以通过构造函数传递该任务
-
outcome
- 任务运行的结果，可以通过get系列方法来获取该结果
- outcome没有被volatie修饰，是因为变量state已经被volatie修饰了，这里是借用volatile的内存语义来保证写入outcome时会把值刷新到主内存，读取时会从主内存读取，从而避免多线程下内存不可见问题
-
runner变量
- 记录了运行该任务的线程，这个是FutureTask的run方法内使用CAS函数设置的
-
waiters变量
-
是一个WaitNode节点，是用Treiber stack实现的无锁栈，栈顶元素用waiters代表。栈用来记录所有等待任务结果的线程节点
- 该栈式一个简单的联表，用来记录所有等待结果被阻塞的线程
-
-
构造函数
FutureTask(Callable)
-
FutureTask(Runnable, V)
- 是用适配器模式来做转换
-
额外说明
- FutureTask中是用UNSAFE机制来操作内存变量
-
-
-
FutureTask的run()方法
-
说明
- 该方法时任务的执行体，线程时调用该方法来具体运行任务的，如果任务没有被取消，则该方法会运行任务，并且将结果设置到outcome变量中
- 如果直接运行，则不是在单独线程中主席那个
源码
-
-
FutureTask的get()方法
-
说明
- 等待异步计算任务完成，并返回结果；如果当前任务计算还没完成则会阻塞调用线程直到任务完成；在等待结果的过程中可能出现CancellationException、InterruptedException、ExecutionException异常
源码
awaitDone方法的实现
get(long,TimeUnit)
-
-
FutureTask的cancel(boolean)方法
-
说明
- 尝试取消任务的执行。传true时会中断正在主席那个任务的线程
源码
-
-
总结
- 当我们创建一个FutureTask时，其任务状态初始化为NEW，当我们把任务提交到线程或者线程池后，会有一个线程来执行该FutureTask任务，具体是调用其run方法执行任务。在任务执行过程中，可以在其他线程调用FutureTask的get()方法来等待获取结果，如果当前任务还在执行，则调用get的线程会被阻塞然后放入FutureTask内的阻塞链表队列；毒功而线程可以同时调用get方法，这些可能都会被阻塞并放到阻塞链表队列中。当任务主席那个完毕后会把结果或者异常信息设置到outcome变量，然后会移除和唤醒FutureTask内阻塞链表队列中的线程节点，进而这些由于FutureTask的get方法而被阻塞的线程就会被激活
-
FutureTask的局限性
-
列举
- 并发代码不简洁
- 不能表达多个FutureTask之间的关系
- get方法是阻塞式
-
要求
- 结合多个异步计算，比如独立或者依赖
- 对反应式编程的支持，也就是当任务计算完成后能进行通知，并且可以以计算结果作为一个动作的参数进行下一步计算
- 可以退共国编程的方式手动设置（代码的方式）Future的结果；FutureTask不能实现让用户通过函数来设置其结果，而是在其任务内部来设置
- 可以等多个Future对应的计算结果都出来后做一些事情
-
解决
- JDK8的CompletableFuture
-
JDK中的CompletableFuture
-
概述
-
简介
- CompletableFuture是一个可以通过编程方式设置借宿那结果和状态以便让任务结束的Future，并且其可以作为一个CompletionState，当它的计算完成时可以出发一个函数或者行为；当毒功而线程企图调用同一个CompletableFuture的complete、cancel方式时只有一个线程会成功。
-
额外说明
-
CompletableFuture还实现了CompletionState的一些接口
- 当CompletableFuture任务完成后，同步是用任务执行线程来执行依赖任务结果的函数或者行为
- 所有异步的方法在没有显示指定Esxecutro参数的情况下都是复用ForkJoinPool.commonPool()线程池来执行
- 所有CompletionState方法的实现都是相互独立的，以便一个方法的行为不会因为重载了其他方法而受影响
一个ComletableFuture任务可能有一些依赖其计算结果的行为方法，这些行为方法被手机到一个无所基于CAS操作的链接起来的链表组成的栈中；当CompletableFuture的计算任务完成后，会自动弹出栈中的行为方法并执行
-
-
ForkJoinPool框架
与其他ExecutorService的区别是是用了工作窃取算法来提供性能，其内部每个线程都关联自己的内存队列，正常情况下每个线程从自己队列里面获取并执行，当本身队列没有任务时，当前线程会去其他线程概念来奶的队列里面获取任务来执行。
是用与很多任务会产生子任务或者由很多小任务的情况
commonPool静态线程池
-
commonPool的参数可以通过system properties中的三个参数来控制
-
java.util.concurrent.ForkJoinPool.common.parallelism
- 并行度级别，非负整数
-
java.util.concurrent.ForkJoinPool.common.threadFactory
- ForkJoinWorkerThreadFactory的类名
-
java.util.concurrent.ForkJoinPool.common.exceptionHanderl
- UncaughtExceptionHanderl的类名
-
可以使用构造函数显示设置线程格式，默认时当前机器上可用的CPU个数
提供了任务执行、任务生命周期控制的方法、任务状态检测的方法、toString可以返回当前线程池的状态
当线程池关闭或者内部资源被耗尽，再向线程池提交任务会抛出RejectedExecutionException异常
-
显示设置CompletableFuture结果
-
基于CompletableFuture实现异步计算与结果转换
-
基于runAsync系列方法实现无返回值的异步计算
- 打印日志、异步做消息通知等
- 可以简化编程
- runAsync(Runnable runnable)方法是使用震哥哥JVM内唯一的ForkJoinPool.commonPoool()线程池来执行异步任务的，使用RunAsync(Runnable,Executor)方法允许我们使用自定义的线程池来执行异步任务
-
基于supplyAsync系列方法实现由返回值的异步计算
- 默认线程池同上
-
基于thenRun实现异步任务A，执行完毕后，激活异步任务B执行
- 可以使用thenAcceptAsync来指定设置的回调事件使用自定义线程池来主席那个
thenApply，同上，但是任务B可以拿到任务A的执行结果
-
whenComplete设置回调函数，当异步任务执行完毕后进行回调，不会阻塞调用线程
- 测试时要挂起主线程，等待异步让你无执行完毕
-
总结
- 使用CompletableFuture时，大多数时候不需要显示拆功能键线程池，并传递任务到线程池内，简化编程负担，也可以使用自定义线程池
-
-
多个CompletableFuture进行组合运算
- thenCompose，一个CompletableFuture执行完毕后，执行另外一个CompletableFuture
- thenCombine实现当来给你个并发运行的CompletableFuture任务都完成后，使用两者的结果作为参数再执行一个异步任务
- allOf等待多个并发运行的CompletableFuture任务执行完毕
- anyOf等多个并发运行的CompletableFuture任务中有一个执行完毕
-
异常处理
- CompletableFuture提供了completeExceptionally方法来处理异常情况
CompletableFuture概要原理
JDK8 StreamCompletableFuture
- JDK8 Stream
- 当Stream遇见CompletableFuture
总结
Spring框架中的异步执行
Spring中对TaskExecutor的抽象
如何在Spring中使用异步执行
- 使用TaskExecutor实现异步执行
- 使用注解@Async实现异步执行
@Async注解异步执行原理
总结
基于反应式编程实现异步编程
反应式编程概述
Reactive Stream规范
基于RxJava实现异步编程
基于Reactor实现异步编程
总结
Web Servlet的异步非阻塞处理
Servlet概述
Servlet 3.0提供的异步处理能力
Servlet 3.1提供的非阻塞IO能力
Spring Web MVC的异步处理能力
- 基于DeferredResult的异步处理
- 基于Callable实现异步处理
总结
Spring WebFlux的异步非阻塞处理
Spring WebFlux概述
Reactive编程Reactor库
WebFlux服务器
WebFlux的编程模型
WebFlux的编程模型
- WebFlux注解式编程模型
- WebFlux函数式编程模型
WebFlux原理浅尝
- Reactor Netty概述
- WebFlux服务器启动流程
- WebFlux一次服务调用流程
WebFlux的使用场景
总结
高性能异步编程框架和中间件
异步、基于事件驱动的网络编程框架——Netty
- Netty概述
- Netty的线程模型
- TCP半包与粘包问题
高性能RPC框架——Apache Dubbo
- Apache Dubbo概述
- Dubbo的异步调用
- Dubbo的异步执行
高性能线程间消息传递库——Disruptor
- Disruptor概述
- Disruptor的特性详解
- 基于Disruptor实现异步编程
异步、分布式、基于消息驱动的框架——Akka
- Akka概述
- 传统编程模型存在的问题
- Actor模型解决了传统编程模型的问题
- 基于Akka实现异步编程
高性能分布式消息框架——Apache RocketMQ
- Apache RocketMQ概述
- 基于Apache RocketMQ实现系统间异步解耦
总结
Go语言的异步编程能力
Go语言概述
Go语言的线程模型
- 一对一模型
- 多对一模型
- 多对多模型
- Go语言的线程模型
goroutine与channel
- goroutine
- channel
- 构建管道实现异步编程
