webFlux&Reactor

时间：2020-03-22 19:44:56 阅读：166 评论：0 收藏：0 [点我收藏+]

标签：print handler col configure 大量非阻塞 route active ppi

　　配置springcloud的gateway的时候，需要用到webflux，所以需要学习一下。以下是目前我的理解，可能不正确，但是会持续修正。

　　什么是webflux？目前的认知是异步非阻塞IO的webMVC，因为之前的Springmvc是基于同步阻塞IO模型的Servlet实现的，包括tomcat，jetty等传统的servlet容器，因为他们的servlet不支持异步非阻塞，所以，每个请求在获取资源的时候，系统资源都在该请求的名下，显而易见，这是会浪费很多资源的，因为在进行资源IO的时候，如果资源阻塞，线程等资源需要等待被请求资源的IO唤醒。所以传统的tomcat，jetty在性能方面有很大的限制，这也是springmvc面临的问题。

　　所以springwebflux伴随spring5出现了，基于servlet3.1支持异步非阻塞IO，也就是说当请求a的资源IO出现阻塞并不会影响b资源的请求。

以及下图sringboot2.0中webflux和mvc技术栈的对比：

　　技术图片

　　可以看出响应式Reactive技术栈是其新的技术要点。

Router Functions: 对应@Controller，@RequestMapping等标准的Spring MVC注解，提供一套函数式风格的API，用于创建Router，Handler和Filter。

Webflux：核心模块，联系Router Functions模块和Reactive Streams模块，提供响应式编程。

Reactive Streams：一种支持背压（Backpressure）机制的异步数据流处理标准，主要实现有RxJava和Reactor，springwebflux采用的是Reactor。

响应式编程：

　　响应式编程就是基于reactor的思想，当你做一个带有一定延迟的才能够返回的io操作时，不会阻塞，而是立刻返回一个流，并且订阅这个流，当这个流上产生了返回数据，可以立刻得到通知并调用回调函数处理数据。

Backpressure机制：

　　生产速度大于消费速度，所以需要 Buffer；

　　外部条件有限制，所以 Buffer 需要有上限；

　　Buffer 达到上限这个现象，有一个简化的等价词叫做 Backpressure；

　　Backpressure 的出现其实是一种危险边界，唯一的选择是丢弃新事件。

Reactor：

　　Reactor模式也叫反应器模式，大多数IO组件如reids，netty都在使用的IO模式，以此来解决高性能并发。Reactor被分为handler和reactor两个部分，前者负责业务处理，后者负责io接受分发。

　　回顾一下IO历史：

　　　　连轴转：一个while处理全部请求（单线程）。一个请求阻塞全部阻塞。

　　　　connection per thread：一个请求一个线程。一个线程只能处理一个请求，即使语法上允许一个线程处理多个请求，但是一个线程上的一个请求被阻塞，其他也会阻塞；每个线程都是系统的资源，耗费资源巨大，创建销毁线程也需要消耗资源。

　　　　Java.NIO: （NIO的Selector网络通讯就是一个单线程版的Reactor）。一个典型的NIO代码：

 1 static class Server
 2     {
 3 
 4         public static void testServer() throws IOException
 5         {
 6 
 7             // 1、获取Selector选择器
 8             Selector selector = Selector.open();
 9 
10             // 2、获取通道
11             ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
12             // 3.设置为非阻塞
13             serverSocketChannel.configureBlocking(false);
14             // 4、绑定连接
15             serverSocketChannel.bind(new InetSocketAddress(SystemConfig.SOCKET_SERVER_PORT));
16 
17             // 5、将通道注册到选择器上,并注册的操作为：“接收”操作
18             serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
19 
20             // 6、采用轮询的方式，查询获取“准备就绪”的注册过的操作
21             while (selector.select() > 0)
22             {
23                 // 7、获取当前选择器中所有注册的选择键（“已经准备就绪的操作”）
24                 Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
25                 while (selectedKeys.hasNext())
26                 {
27                     // 8、获取“准备就绪”的时间
28                     SelectionKey selectedKey = selectedKeys.next();
29 
30                     // 9、判断key是具体的什么事件
31                     if (selectedKey.isAcceptable())
32                     {
33                         // 10、若接受的事件是“接收就绪” 操作,就获取客户端连接
34                         SocketChannel socketChannel = serverSocketChannel.accept();
35                         // 11、切换为非阻塞模式
36                         socketChannel.configureBlocking(false);
37                         // 12、将该通道注册到selector选择器上
38                         socketChannel.register(selector, SelectionKey.OP_READ);
39                     }
40                     else if (selectedKey.isReadable())
41                     {
42                         // 13、获取该选择器上的“读就绪”状态的通道
43                         SocketChannel socketChannel = (SocketChannel) selectedKey.channel();
44 
45                         // 14、读取数据
46                         ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
47                         int length = 0;
48                         while ((length = socketChannel.read(byteBuffer)) != -1)
49                         {
50                             byteBuffer.flip();
51                             System.out.println(new String(byteBuffer.array(), 0, length));
52                             byteBuffer.clear();
53                         }
54                         socketChannel.close();
55                     }
56 
57                     // 15、移除选择键
58                     selectedKeys.remove();
59                 }
60             }
61 
62             // 7、关闭连接
63             serverSocketChannel.close();
64         }
65 
66         public static void main(String[] args) throws IOException
67         {
68             testServer();
69         }
70     }