Project Reactor 深度解析随记

悠悠 2022-09-12 14:43 16阅读 0赞

[一、Project Reactor 深度解析 - 1. 响应式编程介绍，实现以及现有问题][Project Reactor _ - 1.]

随记：

1、响应式编程框架：RxJava 、 Project Reactor、Vert.x 以及 java 9 引入的响应式编程的标准接口Flow

2、什么是响应式编程

1）、传统编程存在两个问题：一个问题是：一个请求一个线程，是阻塞式调用，高并发时，如果一个IO线程阻塞的时间比较长，有可能导致线程池所有的线程都处于阻塞状态，这样就没法处理请求了。另一个问题是：一个请求要并行处理时，写代码比较麻烦。  
2）、基于回调的Callback接口（java 8 引入的各种Functional Interface),另一种是Future 框架。

//获取回答
    public void getAnswer(Connection connection, HttpRequest request) { 
        ResultVO resultVO = new ResultVO();
        getVoteCountFromCache(id, (count, throwable) -> { 
            //异常不为null则为获取失败
            if (throwable != null) { 
                //读取缓存失败就从数据库获取
                getVoteCountFromDB(id, (count2, throwable2) -> { 
                    if (throwable2 == null) { 
                        resultVO.setVoteCount(voteCount);
                    }
                    //从数据库读取回答信息
                    getAnswerFromDB(id, (answer, throwable3) -> { 
                        if (throwable3 == null) { 
                            resultVO.setAnswer(answer);
                            connection.write(resultVO);
                        } else { 
                            connection.write(throwable3);
                        }
                    });
                });
            } else { 
                //获取成功，设置voteCount
                resultVO.setVoteCount(voteCount);
                //从数据库读取回答信息
                getAnswerFromDB(id, (answer, throwable2) -> { 
                    if (throwable2 == null) { 
                        resultVO.setAnswer(answer);
                        //返回响应
                        connection.write(resultVO);
                    } else { 
                        //返回错误响应
                        connection.write(throwable2);
                    }
                });
            }
        });
    }

callback方式，调用层次越深，代码越难写，重复的代码越多，并发也是很难写的。

Future 方式：这种方式代码简洁了，而且也可以并行了。

//获取回答
    public void getAnswer(Connection connection, HttpRequest request) { 
        ResultVO resultVO = new ResultVO();
            //所有的异步任务都执行完之后要做的事情
            CompletableFuture.allOf(
                    getVoteCountFromCache(id)
                            //发生异常，从数据库读取
                            .exceptionallyComposeAsync(throwable -> getVoteCountFromDB(id))
                            //读取完之后，设置VoteCount
                            .thenAccept(voteCount -> { 
                        resultVO.setVoteCount(voteCount);
                    }),
                    getAnswerFromDB(id).thenAccept(answer -> { 
                        resultVO.setAnswer(answer);
                    })
            ).exceptionallyAsync(throwable -> { 
                connection.write(throwable);
            }).thenRun(() -> { 
                connection.write(resultVO);
            });
    }

所以，响应式编程的本质是：IO线程将数据写入 Connection 之后，将响应返回后需要做的事情以及参数缓存到一个地方之后，直接返回。在有响应返回后，NIO 的 Selector 的 Read 事件会是 Ready 状态，扫描 Selector 事件的线程，会告诉你的线程池数据好了，然后线程池中的某个线程，拿出刚刚缓存的要做的事情还有参数，继续处理。  
3、背压（Back Pressure）问题：背压是指，当上游请求过多，下游服务来不及响应，导致 Buffer 溢出的这样一个问题。在响应式编程，由于线程不阻塞，遇到 IO 就会把当前参数和要做的事情缓存起来，这样无疑增大了很多吞吐量，同时内存占用也大了起来，如果不限制的话，很可能 OutOfMemory，这就是背压问题。

4、Project Reactor 在 Completableuture 这种实现的基础上，增加了更多的组合方式以及更完善的异常处理机制，以及面对背压时候的处理机制，还有重试机制。

[二、Project Reactor 深度解析 - 2. 响应式编程调试，FLow的概念设计以及实现][Project Reactor _ - 2. _FLow]  
1、如何调试，启动参数中加入 -Dreactor.trace.operatorStacktrace=true，这样启动全局 Operator 堆栈追踪，或者在代码中找开或者关闭：

//打开
    Hooks.onOperatorDebug();
    //关闭
    Hooks.resetOnOperatorDebug();

2、Project Reactor - Flux如何实现Flow的接口

[三、响应式编程 & Project Reactor 概述][_ Project Reactor]

1、Project Reactor 是 响应式编程 (Reactive Programming) 理论的实现  
2、Data streams: 数据流. 分为 静态数据流 (数组, 文件等) 和 动态数据流 (事件流, 日志流等).  
3、响应式编程的特性

*  描述而非执行: 在最终调用 subscribe() 方法之前, 从发布端到订阅端没有任何事情发生. 就好比无论多长的水管. 只要水龙头不拧开, 水管里的水就不会流动. 为了提高描述能力, 响应式编程提供了比 Java 8 的 Stream API 多得多的 API, 如 buffer(), merge(), orErrorMap() 等
 *  提高吞吐量: 类似 HTTP/2 中的连接复用, 响应式编程通过线程复用来提高吞吐量. 在传统的 Servlet 容器中, 每进来一个请求就会申请一个线程来处理. 受限于机器硬件资源, 单台服务器所能支撑的线程数存在极限的. 假设为 T, 那么应用同时能处理的请求数 (吞吐量) 必然也不会超过 T. 但是对于一个使用 Spring 5 开发的基于响应式编程框架的应用, 通过线程复用技术, 最大吞吐量是有可能超过 T 的.
 *  Backpressure (背压) 支持: 简单来说, 背压就是一种反馈机制. 在一般的 Push 模型中, 发布者既不知道也不关心订阅者的处理速度, 当数据的发布速度超过处理速度时, 需要订阅者自己决定缓存还是丢弃. 如果使用响应式编程框架, 决定权就交回给发布者, 订阅者只需要根据自己的处理能力向发布者请求相应数量的数据. 你可能会问这不就是 Pull 模型嘛? 其实是不同的: 在Pull模型中, 订阅者每次处理完数据, 都要重新发起一次请求拉取新的数据. 使用背压, 订阅者只需要发起一次请求, 就能连续不断的重复请求数据.

4、Best practices

*  一次请求涉及多次外部服务调用
 *  非可靠的网络传输
 *  高并发下的消息处理
 *  弹性计算网络  
    5、代价
 *  虽然复用线程有助于提高吞吐量, 但是一旦在某个回调函数中线程被卡住, 那么这个想成上的所有请求都会被阻塞. 最严重的的情况是整个应用会被拖垮.
 *  难以调试. 由于响应式编程 API 的强大的描述能力, 在一个典型应用中, 大部分代码都是以链式表达式的形式构成, 将很难定位到具体哪个环节出了问题. 需要专门的调试方法和工具 (以后介绍).

[Project Reactor _ - 1.]: https://zhanghaoxin.blog.csdn.net/article/details/107674092
[Project Reactor _ - 2. _FLow]: https://zhanghaoxin.blog.csdn.net/article/details/107771911
[_ Project Reactor]: https://blog.csdn.net/caplike/article/details/108063454