Reactive 简介
1. 概念
Reactive 非常适合低延迟、高吞吐量的工作负载。
Reactive Processing 是一种范式（规范），它使开发人员能够构建非阻塞的、异步的应用程序，这些应用程序能够处理背压（流控制）
Reactive Streams 为无阻塞背压的异步流处理提供标准。
Reactor 是基于Reactive Streams规范的第四代响应库，用于在JVM上构建非阻塞的应用程序。
Project Reactor 是一个完全无阻塞的基础，其中包括背压支持。它是Spring生态系统中的响应式堆栈的基础，并且在诸如Spring WebFlux，Spring Data和Spring Cloud Gateway等项目中都有它的身影。利用Project Reactor可以高效的响应式系统。刚才说Reactive Streams是规范，那么Project Reactor就是实现。
2. 响应式编程
响应式编程是一种异步编程风格，它关注数据流和变化的传播。
响应式编程是一种与数据流和变化传播相关的声明式编程范式。使用此范例，可以轻松地表示静态(例如，数组)或动态(例如，事件发射器)数据流，并且还可以表示关联执行模型中的推断出的依赖关系，这有助于更改数据流的自动传播。
reactive programming （响应式编程）
imperative programming（命令式编程）
在命令式编程中，a:=b+c意味着将b+c的结果赋值给a，并且此后b或c的值发生变化不会影响到a的值。而在响应式编程中，a的值会随着b或c的改变而自动更新，并且不需要重新执行a:=b+c来确定当前分配给a的值。（PS：是不是很像angularjs、vuejs这种MVVM框架，视图绑定模型，模型变了，视图自动就跟着变了）
例如，在 model–view–controller (MVC) 架构中，响应式编程可以促进基础模型中的更改，这些更改会自动反映在关联的视图中。
响应式编程与面向对象编程中通常使用的观察者模式具有很多相似之处。
如果从推拉的角度来看的话，响应式编程是“推”，它主动将变化推送给它的订阅者。Publisher-Subscriber是两个非常重要的概念。
想象一下，数据流从源出发，经过一个一个节点的处理，最终达到目的地。节点就相当于操作符，处理完了以后就将流发射出去，到下一个节点再执行再发射。
我总觉得这个流程很眼熟，很像 Apache Storm 的处理方式。在一个拓扑结构中，数据流从Spout发出，经过若干bolt的处理，最终汇集到某个地方。
还有一种理解，我觉得也很不错，说响应式编程是一种通过异步和数据流来构建事务关系的编程模型。事物可以理解程一次处理过程，一次执行过程。响应式编程就是要构建关系，事务和事务之间的关系。而数据流就像是一个桥梁一样，数据流从一个事务流向下一个事务。
想象一下，长江流经宜宾、泸州、重庆、涪陵、万州、宜昌、荆州、武汉、黄石、鄂州、九江、安庆、铜陵、芜湖、南京、上海，最终汇入东海。
就像CompleteFuture把Future进行编排一样。
本质来讲，响应式编程上是对数据流或某种变化所作出的反应，但是这个变化什么时候发生是未知的，所以他是一种基于异步、回调的方式在处理问题
3. NIO
NIO（Non-Blocking I/O）
BIO（Blocking I/O）
在经典的线程模型中，socket.accept()、socket.read()、socket.write()三个主要函数都是同步阻塞的，当一个连接在处理I/O的时候，系统是阻塞的，如果使用单线程的话就阻塞在那里了，但CPU是并没有阻塞，如果用多线程的话，就可以让CPU去处理更多的事情。其实这也是所有使用多线程的本质： 当I/O阻塞系统，但CPU空闲的时候，可以利用多线程使用CPU资源。然而，线程的创建、销毁、切换成本都是很高的。
事实上，所有的系统I/O都分为两个阶段：等待就绪和操作。举例来说，读函数，分为等待系统可读和真正的读；同理，写函数分为等待网卡可以写和真正的写。
需要说明的是等待就绪的阻塞是不使用CPU的，是在“空等”；而真正的读写操作的阻塞是使用CPU的，真正在”干活”。
以socket.read()为例子：
传统的BIO里面socket.read()，如果TCP RecvBuffer里没有数据，函数会一直阻塞，直到收到数据，返回读到的数据。
对于NIO，如果TCP RecvBuffer有数据，就把数据从网卡读到内存，并且返回给用户；反之则直接返回0，永远不会阻塞。
在BIO模型中，没有办法知道到底能不能写、能不能读，只能”傻等”。而在NIO模型中，如果一个连接不能读写（socket.read()返回0或者socket.write()返回0），我们可以把这件事记下来，记录的方式通常是在Selector上注册标记位，然后切换到其它就绪的连接（channel）继续进行读写。
NIO的主要事件有几个：读就绪、写就绪、有新连接到来。那么，首先需要注册当这几个事件到来的时候所对应的处理器，然后在合适的时机告诉事件选择器：我对这个事件感兴趣，最后用一个死循环选择就绪的事件。select是阻塞的，所以你可以放心大胆地在一个while(true)里面调用这个函数而不用担心CPU空转。
总结起来就是：注册所有感兴趣的事件处理器，单线程轮询选择就绪事件，执行事件处理器。
我们大概可以总结出NIO是怎么解决掉线程的瓶颈并处理海量连接的：
NIO由原来的阻塞读写（占用线程）变成了单线程轮询事件，找到可以进行读写的网络描述符进行读写。除了事件的轮询是阻塞的（没有可干的事情必须要阻塞），剩余的I/O操作都是纯CPU操作，没有必要开启多线程。
NIO由原来的阻塞读写（占用线程）变成了单线程轮询事件，找到可以进行读写的网络描述符进行读写。除了事件的轮询是阻塞的（没有可干的事情必须要阻塞），剩余的I/O操作都是纯CPU操作，没有必要开启多线程。并且由于线程的节约，连接数大的时候因为线程切换带来的问题也随之解决，进而为处理海量连接提供了可能。单线程处理I/O的效率确实非常高，没有线程切换，只是拼命的读、写、选择事件。但现在的服务器，一般都是多核处理器，如果能够利用多核心进行I/O，无疑对效率会有更大的提高。
Buffer（缓冲区）
在NIO中，所有数据都是用缓冲区处理的。在读取数据时，它是直接读到缓冲区中的；在写入数据时，它也是写入到缓冲区中的。
Channel（通道）
通道是一个对象，通过它可以读取和写入数据，当然了所有数据都通过Buffer对象来处理。我们永远不会将字节直接写入通道中，相反是将数据写入包含一个或者多个字节的缓冲区。同样不会直接从通道中读取字节，而是将数据从通道读入缓冲区，再从缓冲区获取这个字节。
Selector（选择器）
Selector类是NIO的核心类，Selector（选择器）选择器提供了选择已经就绪的任务的能力。Selector会不断的轮询注册在上面的所有channel，如果某个channel为读写等事件做好准备，那么就处于就绪状态，通过Selector可以不断轮询发现出就绪的channel，进行后续的IO操作。一个Selector能够同时轮询多个channel。这样，一个单独的线程就可以管理多个channel，从而管理多个网络连接。这样就不用为每一个连接都创建一个线程，同时也避免了多线程之间上下文切换导致的开销。
一个简单的读取文件的例子：
1 package com.cjs.example.restservice.nio;2 3 import java.io.FileInputStream;4 import java.nio.ByteBuffer;5 import java.nio.channels.FileChannel;6 7 /**8 * @author ChengJianSheng9 * @date 2020-03-26
10 */
11 public class Hello {12
13 public static void main(String[] args) throws Exception {14 FileInputStream fis = new FileInputStream("/data.txt");
15 FileChannel channel = fis.getChannel();
16
17 ByteBuffer buffer = ByteBuffer.allocate(10);
18
19 while (true) {20 if (channel.read(buffer) == -1) {21 break;
22 }
23 buffer.flip();
24 while (buffer.hasRemaining()) {25 System.out.print((char)buffer.get());
26 }
27 buffer.clear();
28 }
29
30 channel.close();
31 fis.close();
32 }
33 }
Server.java
1 package com.cjs.example.restservice.nio;2 3 import java.net.InetSocketAddress;4 import java.nio.ByteBuffer;5 import java.nio.channels.SelectionKey;6 import java.nio.channels.Selector;7 import java.nio.channels.ServerSocketChannel;8 import java.nio.channels.SocketChannel;9 import java.util.Iterator;
10 import java.util.Set;
11
12 /**
13 * @author ChengJianSheng
14 * @date 2020-03-26
15 */
16 public class Server {17 public static void main(String[] args) throws Exception {18 // 创建一个Selector
19 Selector selector = Selector.open();
20
21 ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
22 serverSocketChannel.configureBlocking(false);
23 serverSocketChannel.bind(new InetSocketAddress(9000));
24
25 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
26
27 while (true) {28 selector.select();
29
30 Set<SelectionKey> selectedKeys = selector.selectedKeys();
31 Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
32 while (keyIterator.hasNext()) {33 SelectionKey key = keyIterator.next();
34 if(key.isAcceptable()) {35 // a connection was accepted by a ServerSocketChannel.
36
37 System.out.println(1);
38 ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
39 SocketChannel sc = ssc.accept();
40 sc.configureBlocking(false);
41 sc.register(selector, SelectionKey.OP_READ);
42 } else if (key.isConnectable()) {43 // a connection was established with a remote server.
44 } else if (key.isReadable()) {45 // a channel is ready for reading
46
47 System.out.println(2);
48 SocketChannel socketChannel = (SocketChannel) key.channel();
49 ByteBuffer buffer = ByteBuffer.allocate(1024);
50 int len = 0;
51 while ((len = socketChannel.read(buffer)) != -1) {52 buffer.flip();
53 System.out.println(new String(buffer.array(), 0, len));
54 }
55
56 socketChannel.close();
57 } else if (key.isWritable()) {58 // a channel is ready for writing
59 }
60
61 keyIterator.remove();
62 }
63 }
64 }
65 }
Client.java
1 package com.cjs.example.restservice.nio;2 3 import java.net.InetSocketAddress;4 import java.nio.ByteBuffer;5 import java.nio.channels.SocketChannel;6 7 /**8 * @author ChengJianSheng9 * @date 2020-03-26
10 */
11 public class Client {12
13 public static void main(String[] args) throws Exception {14 SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9000));
15 socketChannel.configureBlocking(false);
16
17 ByteBuffer buffer = ByteBuffer.allocate(1024);
18 String msg = "Hello, World!";
19 buffer.put(msg.getBytes());
20 buffer.flip();
21 socketChannel.write(buffer);
22
23 socketChannel.close();
24 }
25 }
关于Selector的用法
1 Selector selector = Selector.open();2 3 channel.configureBlocking(false);4 5 SelectionKey key = channel.register(selector, SelectionKey.OP_READ);6 7 while(true) {8 9 int readyChannels = selector.selectNow();
10
11 if(readyChannels == 0) continue;
12
13
14 Set<SelectionKey> selectedKeys = selector.selectedKeys();
15
16 Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
17
18 while(keyIterator.hasNext()) {19
20 SelectionKey key = keyIterator.next();
21
22 if(key.isAcceptable()) {23 // a connection was accepted by a ServerSocketChannel.
24
25 } else if (key.isConnectable()) {26 // a connection was established with a remote server.
27
28 } else if (key.isReadable()) {29 // a channel is ready for reading
30
31 } else if (key.isWritable()) {32 // a channel is ready for writing
33 }
34
35 keyIterator.remove();
36 }
37 }
参考：
https://spring.io/reactive
https://www.jianshu.com/p/d47835316016
https://www.cnblogs.com/haimishasha/p/10756448.html
https://tech.meituan.com/2016/11/04/nio.html
墙裂推荐Java NIO教程
http://tutorials.jenkov.com/java-nio/index.html
http://tutorials.jenkov.com/java-nio/selectors.html
http://tutorials.jenkov.com/java-nio/server-socket-channel.html
Reactive 简介相关推荐
- Reactive简介
理解Reactive 相关技术 反应堆模式(Reactor) 同步非阻塞,多工模式,一个事情可以分为几个步骤,每个步骤相应去做,同步串行先做A,后做B Proactor模式 异步非阻塞,多工模式,A, ...
- Vert.x和Reactive简介
Vert.x 是一个用于在 JVM 上构建 reactive 应用程序的工具包. 反应式应用程序既可以随着工作负载的增长而可扩展(scalable),又可以在出现故障时可恢复(resilient) ...
- Reactive Extensions简介一
在.Net 4.0中引入了两个新的接口用来实现观察者模式--IObservable和IObserver.IObservable是数据源,IObserver是观察者,观察者订阅数据源后,当新的数据产生时 ...
- redux logic_Redux-Logic简介
redux logic by Sam Ollason 通过萨姆·奥拉森(Sam Ollason) Redux-Logic简介 (An Introduction to Redux-Logic) This ...
- reactor官方文档译文(1)Reactor简介
原文地址:http://projectreactor.io/docs/reference/ Reactor简介 Reactor是一个基础库,用在构建实时数据流应用.要求有容错和低延迟至毫秒.纳秒.皮秒 ...
- 响应式编程简介之:Reactor
文章目录 简介 Reactor简介 reactive programming的发展史 Iterable-Iterator 和Publisher-Subscriber的区别 为什么要使用异步reacti ...
- JDK11的新特性:HTTP API和reactive streams
文章目录 简介 怎么在java中使用reactive streams POST请求的例子 总结 简介 在JDK11的新特性:新的HTTP API中,我们介绍了通过新的HTTP API,我们可以发送同步 ...
- 使用 RxJS 实现 JavaScript 的 Reactive 编程
简介 作为有经验的JavaScript开发者,我们会在代码中采用一定程度的异步代码.我们不断地处理用户的输入请求,也从远程获取数据,或者同时运行耗时的计算任务,所有这些都不能让浏览器崩溃.可以说,这些 ...
- Redux中的功能式React式编程简介
by Bhuvan Malik 通过布凡·马利克(Bhuvan Malik) Redux中的功能式React式编程简介 (An introduction to functional Reactive ...
最新文章
- Python压缩目录文件夹，解压目录文件夹及耗时效率统计
- TVM: Deep Learning模型的优化编译器(强烈推荐, 附踩坑记录)
- python协程学习——写个并发获取网站标题的工具
- 自己动手制作(DIY)一个Mini-Linux系统
- Android http Request / Response ContentType
- Boost锁~临界区保护和临界资源共享
- js矢量图类库：Raphaël—JavaScript Library
- django图片上传到oss_django 配置阿里云OSS存储media文件的例子
- Machine Learning - Coursera week6 Evaluating a learning algorithm
- Spring Boot笔记-解决前后端分离在开发时的跨域问题
- 基于Bounding Box的激光点云聚类
- linux 计算标准差,Azure Linux VM 的计算基准测试分数 - Azure Virtual Machines | Microsoft Docs...
- 不要再被Python洗脑了，来看看这个吧......
- lfm雷达信号处理matlab,雷达信号处理-线性调频（LFM）Chirp信号脉冲压缩
- matlab卷积不从0开始,matlab 编辑卷积出错
- noob的python学习之路
- linux系统编译dfu
- POJ1436Horizontally Visible Segments线段树+lazy
- 获取crumbIssuer
- YOLOv5实现目标识别全流程【超级详细！】