redis为什么这么快、底层磁盘以及IO模型

标签：Reactor redis 线程内核 IO 磁盘 public channel 客户端

一、底层磁盘IO机制

Redis是单进程单线程？为什么这么快：

　　进入redis 安装目录下执行以下命令，查看set lpush命令的处理效率：./redis-benchmark -t set,lpush -n 100000 -q

　　根据官方的数据，Redis 的 QPS 可以达到 10 万左右（每秒请求数）。

　　就我这个虚拟机的性能，可以看到每秒钟处理 13 万多次 set 请求，每秒钟处理 13 万次左右 set 请求。可见redis性能之高.原因有如下三点：

纯内存结构。KV 结构的内存数据库，时间复杂度 O(1)。
单线程。单线程有什么好处呢？
1. 没有创建线程、销毁线程带来的消耗。
2. 避免了上线文切换导致的 CPU 消耗。
3. 避免了线程之间带来的竞争问题，例如加锁释放锁死锁等等。
多路复用。异步非阻塞 I/O，多路复用处理并发连接。

　　Redis采用了一种非常简单的做法，单线程来处理来自所有客户端的并发请求，Redis把任务封闭在一个线程中从而避免了线程安全问题；redis为什么是单线程？官方的解释是，CPU并不是Redis的瓶颈所在，Redis的瓶颈主要在机器的内存和网络的带宽。那么Redis能不能处理高并发请求呢？当然是可以的，至于怎么实现的，我们来具体了解一下。【注意并发不等于并行，并发性I/O流，意味着能够让一个计算单元来处理来自多个客户端的流请求。并行性，意味着服务器能够同时执行几个事情，具有多个计算单元】

单线程为什么这么快？

　　因为 Redis 是基于内存的操作，我们先从内存开始说起。

虚拟存储器（虚拟内存 l Vitual Memory ）：

　　名词解释：主存：内存；辅存：磁盘（硬盘）

　　计算机主存（内存）可看作一个由 M 个连续的字节大小的单元组成的数组，每个字节有一个唯一的地址，这个地址叫做物理地址（PA）。早期的计算机中，如果 CPU 需要内存，使用物理寻址，直接访问主存储器。

　　这种方式有几个弊端：

在多用户多任务操作系统中，所有的进程共享主存，如果每个进程都独占一块物理地址空间，主存很快就会被用完。我们希望在不同的时刻，不同的进程可以共用同一块物理地址空间。
如果所有进程都是直接访问物理内存，那么一个进程就可以修改其他进程的内存数据，导致物理地址空间被破坏，程序运行就会出现异常。

　　为了解决这些问题，我们就想了一个办法，在 CPU 和主存之间增加一个中间层。CPU不再使用物理地址访问，而是访问一个虚拟地址，由这个中间层把地址转换成物理地址，最终获得数据。这个中间层就叫做虚拟存储器（Virtual Memory）。具体的操作如下所示：

　　在每一个进程开始创建的时候，都会分配一段虚拟地址，然后通过虚拟地址和物理地址的映射来获取真实数据，这样进程就不会直接接触到物理地址，甚至不知道自己调用的哪块物理地址的数据。

　　目前，大多数操作系统都使用了虚拟内存，如 Windows 系统的虚拟内存、Linux 系统的交换空间等等。Windows 的虚拟内存（pagefile.sys）是磁盘空间的一部分。在 32 位的系统上，虚拟地址空间大小是 2^32bit=4G。在 64 位系统上，最大虚拟地址空间大小是多少？是不是 2^64bit=1024*1014TB=1024PB=16EB？实际上没有用到 64 位，因为用不到这么大的空间，而且会造成很大的系统开销。Linux 一般用低48 位来表示虚拟地址空间，也就是 2^48bit=256T。可以通过命令cat /proc/cpuinfo查看

　　address sizes : 40 bits physical, 48 bits virtual

　　实际的物理内存可能远远小于虚拟内存的大小。

　　总结：引入虚拟内存，可以提供更大的地址空间，并且地址空间是连续的，使得程序编写、链接更加简单。并且可以对物理内存进行隔离，不同的进程操作互不影响。还可以通过把同一块物理内存映射到不同的虚拟地址空间实现内存共享。

用户空间和内核空间：

　　为了避免用户进程直接操作内核，保证内核安全，操作系统将虚拟内存划分为两部分，一部分是内核空间（Kernel-space）/ˈkɜːnl /，一部分是用户空间（User-space）。

　　内核是操作系统的核心，独立于普通的应用程序，可以访问受保护的内存空间，也有访问底层硬件设备的权限。

　　内核空间中存放的是内核代码和数据，而进程的用户空间中存放的是用户程序的代码和数据。不管是内核空间还是用户空间，它们都处于虚拟空间中，都是对物理地址的映射。

　　在 Linux 系统中, 内核进程和用户进程所占的虚拟内存比例是 1:3。

　　当进程运行在内核空间时就处于内核态，而进程运行在用户空间时则处于用户态。进程在内核空间以执行任意命令，调用系统的一切资源；在用户空间只能执行简单的运算，不能直接调用系统资源，必须通过系统接口（又称 system call），才能向内核发出指令。top 命令：

　　us 代表 CPU 消耗在 User space 的时间百分比;

　　sy 代表 CPU 消耗在 Kernel space 的时间百分比。

进程切换（上下文切换）：

　　多任务操作系统是怎么实现运行远大于 CPU 数量的任务个数的？当然，这些任务实际上并不是真的在同时运行，而是因为系统通过时间片分片算法，在很短的时间内，将CPU 轮流分配给它们，造成多任务同时运行的错觉。

　　为了控制进程的执行，内核必须有能力挂起正在 CPU 上运行的进程，并恢复以前挂起的某个进程的执行。这种行为被称为进程切换。

什么叫上下文？

　　在每个任务运行前，CPU 都需要知道任务从哪里加载、又从哪里开始运行，也就是说，需要系统事先帮它设置好 CPU 寄存器和程序计数器(Program Counter)，这个叫做CPU 的上下文。

　　而这些保存下来的上下文，会存储在系统内核中，并在任务重新调度执行时再次加载进来。这样就能保证任务原来的状态不受影响，让任务看起来还是连续运行。在切换上下文的时候，需要完成一系列的工作，这是一个很消耗资源的操作。

进程的阻塞：

　　正在运行的进程由于提出系统服务请求（如 I/O 操作），但因为某种原因未得到操作系统的立即响应，该进程只能把自己变成阻塞状态，等待相应的事件出现后才被唤醒。进程在阻塞状态不占用 CPU 资源。

文件描述符 FD:

　　Linux 系统将所有设备都当作文件来处理，而 Linux 用文件描述符来标识每个文件对象。文件描述符（File Descriptor）是内核为了高效管理已被打开的文件所创建的索引，用于指向被打开的文件，所有执行 I/O 操作的系统调用都通过文件描述符；文件描述符是一个简单的非负整数，用以表明每个被进程打开的文件。Linux 系统里面有三个标准文件描述符。0：标准输入（键盘）；1：标准输出（显示器）；2：标准错误输出（显示器）。

　　Redis的高性能主要依赖于几个方面。 C语言实现，C语言在一定程度上还是比Java语言性能要高一些，因为C语言不需要经过JVM进行翻译。纯内存I/O，内存I/O比磁盘I/O性能更快 I/O多路复用，基于epoll的I/O多路复用技术，实现高吞吐网络I/O 单线程模型，单线程无法利用到多核CPU，但是在Redis中，性能瓶颈并不是在计算上，而是在I/O 能力，所以单线程能够满足高并发的要求。从另一个层面来说，单线程可以避免多线程的频繁上下文切换以及同步锁机制带来的性能开销。下面我们分别从上述几个方面进行展开说明。

从请求处理开始分析：

　　当我们在客户端向Redis Server发送一条指令，并且得到Redis回复的整个过程中，Redis做了什么呢？

　　要处理命令，则redis必须完整地接收客户端的请求，并将命令解析出来，再将结果读出来，通过网络回写到客户端。整个工序分为以下几个部分：

接收，通过TCP接收到命令，可能会历经多次TCP包、ack、IO操作
解析，将命令取出来
执行，到对应的地方将value读出来
返回，将value通过TCP返回给客户端，如果value较大，则IO负荷会更重

　　其中解析和执行是纯cpu/内存操作，而接收和返回主要是IO操作，首先我们先来看通信的过程。

网络IO的通信原理：

　　同样，我也画了一幅图来描述网络数据的传输流程

　　首先，对于TCP通信来说，每个TCP Socket的内核中都有一个发送缓冲区和一个接收缓冲区接收缓冲区把数据缓存到内核，若应用进程一直没有调用Socket的read方法进行读取，那么该数据会一直被缓存在接收缓冲区内。不管进程是否读取Socket，对端发来的数据都会经过内核接收并缓存到 Socket的内核接收缓冲区。

　　read所要做的工作，就是把内核接收缓冲区中的数据复制到应用层用户的Buffer里。进程调用Socket的send发送数据的时候，一般情况下是将数据从应用层用户的Buffer里复制到Socket的内核发送缓冲区，然后send就会在上层返回。换句话说，send返回时，数据不一定会被发送到对端。

　　网卡中的缓冲区既不属于内核空间，也不属于用户空间。它属于硬件缓冲，允许网卡与操作系统之间有个缓冲；内核缓冲区在内核空间，在内存中，用于内核程序，做为读自或写往硬件的数据缓冲区；用户缓冲区在用户空间，在内存中，用于用户程序，做为读自或写往硬件的数据缓冲区网卡芯片收到网络数据会以中断的方式通知CPU，我有数据了，存在我的硬件缓冲里了，来读我啊。 CPU收到这个中断信号后，会调用相应的驱动接口函数从网卡的硬件缓冲里把数据读到内核缓冲区，正常情况下会向上传递给TCP/IP模块一层一层的处理。

BIO 传统阻塞IO ：

　　Redis的通信采用的是多路复用机制，什么是多路复用机制呢？由于Redis是C语言实现，为了简化大家的理解，我们采用Java语言来描述这个过程。在理解多路复用之前，我们先来了解一下BIO。

　　BIO模型在Java中，如果要实现网络通信，我们会采用Socket套接字来完成。

　　Socket这不是一个协议，而是一个通信模型。其实它最初是BSD发明的，主要用来一台电脑的两个进程间通信，然后把它用到了两台电脑的进程间通信。所以，可以把它简单理解为进程间通信，不是什么高级的东西。主要做的事情不就是：

A发包：发请求包给某个已经绑定的端口（所以我们经常会访问这样的地址182.13.15.16:1235， 1235就是端口）；收到B的允许；然后正式发送；发送完了，告诉B要断开链接；收到断开允许，马上断开，然后发送已经断开信息给B。
B收包：绑定端口和IP；然后在这个端口监听；接收到A的请求，发允许给A，并做好接收准备，主要就是清理缓存等待接收新数据；然后正式接收；接受到断开请求，允许断开；确认断开后，继续监听其它请求。

　　可见，Socket其实就是I/O操作，Socket并不仅限于网络通信，在网络通信中，它涵盖了网络层、传输层、会话层、表示层、应用层——其实这都不需要记，因为Socket通信时候用到了IP和端口，仅这两个就表明了它用到了网络层和传输层；而且它无视多台电脑通信的系统差别，所以它涉及了表示层；一般 Socket都是基于一个应用程序的，所以会涉及到会话层和应用层。

　　构建基础的BIO通信模型，BIO有什么弊端呢？当服务端收到客户端的请求后，不直接返回，而是等待20s。

public class BIOServerSocket {
    //先定义一个端口号，这个端口的值是可以自己调整的。
    static final int DEFAULT_PORT = 8080;

    public static void main(String[] args) throws IOException,
            InterruptedException {
        ServerSocket serverSocket = null;
        serverSocket = new ServerSocket(DEFAULT_PORT);
        System.out.println("启动服务，监听端口：" + DEFAULT_PORT);
        while (true) { //case1: 增加循环，允许循环接收请求
            Socket socket = serverSocket.accept();
            System.out.println("客户端：" + socket.getPort() + "已连接");
            BufferedReader bufferedReader = new BufferedReader(new
                    InputStreamReader(socket.getInputStream()));
            String clientStr = bufferedReader.readLine(); //读取一行信息
            System.out.println("客户端发了一段消息：" + clientStr);
            Thread.sleep(20000); //case2: 修改：增加等待时间
            BufferedWriter bufferedWriter = new BufferedWriter(new
                    OutputStreamWriter(socket.getOutputStream()));
            bufferedWriter.write("我已经收到你的消息了\n");
            bufferedWriter.flush(); //清空缓冲区触发消息发送
        }
    }
}

　　这个情况会导致一个问题，如果服务端在同一个时刻只能处理一个客户端的连接，而如果一个网站同时有1000个用户访问，那么剩下的999个用户都需要等待，而这个等待的耗时取决于前面的请求的处理时长.

　　　　基于多线程优化BIO 为了让服务端能够同时处理更多的客户端连接，避免因为某个客户端连接阻塞导致后续请求被阻塞，于是引入多线程技术。如图所示，当引入了多线程之后，每个客户端的链接（Socket），我们可以直接给到线程池去执行，而由于这个过程是异步的，所以并不会同步阻塞影响后续链接的监听，因此在一定程度上可以提升服务端链接的处理数量。

　　传统 I/O 数据拷贝:

　　以读操作为例：当应用程序执行 read 系统调用读取文件描述符（FD）的时候，如果这块数据已经存在于用户进程的页内存中，就直接从内存中读取数据。如果数据不存在，则先将数据从磁盘加载数据到内核缓冲区中，再从内核缓冲区拷贝到用户进程的页内存中。（两次拷贝，两次 user 和 kernel 的上下文切换）。

　　I/O 的阻塞到底阻塞在哪里？Blocking I/O:

　　当使用 read 或 write 对某个文件描述符进行过读写时，如果当前 FD 不可读，系统就不会对其他的操作做出响应。从设备复制数据到内核缓冲区是阻塞的，从内核缓冲区拷贝到用户空间，也是阻塞的，直到 copy complete，内核返回结果，用户进程才解除block 的状态。

　　为了解决阻塞的问题，我们有几个思路。

在服务端创建多个线程或者使用线程池，但是在高并发的情况下需要的线程会很多，系统无法承受，而且创建和释放线程都需要消耗资源。
由请求方定期轮询，在数据准备完毕后再从内核缓存缓冲区复制数据到用户空间（非阻塞式 I/O），这种方式会存在一定的延迟。

　　能不能用一个线程处理多个客户端请求？

NIO非阻塞IO：

　　使用多线程的方式来解决这个问题，仍然有一个缺点，线程的数量取决于硬件配置，所以线程数量是有限的，如果请求量比较大的时候，线程本身会收到限制从而并发量也不会太高。那怎么办呢，我们可以采用非阻塞IO。 NIO 从JDK1.4 提出的，本意是New IO，它的出现为了弥补原本IO的不足，提供了更高效的方式，提出一个通道（channel）的概念，在IO中它始终以流的形式对数据的传输和接受，下面我们演示一下NIO 的使用。　　所谓的NIO（非阻塞IO），其实就是取消了IO阻塞和连接阻塞，当服务端不存在阻塞的时候，就可以不断轮询处理客户端的请求，如图所示，表示NIO下的运行流程。

　　　　上述这种NIO的使用方式，仍然存在一个问题，就是客户端或者服务端需要通过一个线程不断轮询才能获得结果，而这个轮询过程中会浪费线程资源。复制代码

public class NIOServerSocket {
    public static void main(String[] args) {
        try {
            ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false); //设置连接非阻塞
            serverSocketChannel.socket().bind(new InetSocketAddress(8080));
            while(true){
                //是非阻塞的
                SocketChannel socketChannel=serverSocketChannel.accept(); //获得一个客户端连接
//                socketChannel.configureBlocking(false);//IO非阻塞
                if(socketChannel!=null){
                    ByteBuffer byteBuffer=ByteBuffer.allocate(1024);
                    int i=socketChannel.read(byteBuffer);
                    Thread.sleep(10000);
                    byteBuffer.flip(); //反转
                    socketChannel.write(byteBuffer);
                }else{
                    Thread.sleep(1000);
                    System.out.println("连接位就绪");
                }
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

　　大家站在全局的角度再思考一下整个过程，有哪些地方可以优化呢？　　

NIO多路复用机制：

I/O 指的是网络 I/O。
多路指的是多个 TCP 连接（Socket 或 Channel）。
复用指的是复用一个或多个线程。

　　它的基本原理就是不再由应用程序自己监视连接，而是由内核替应用程序监视文件描述符。客户端在操作的时候，会产生具有不同事件类型的 socket。在服务端，I/O 多路复用程序（I/O Multiplexing Module）会把消息放入队列中，然后通过文件事件分派器（Fileevent Dispatcher），转发到不同的事件处理器中。

　　多路复用有很多的实现，以 select 为例，当用户进程调用了多路复用器，进程会被阻塞。内核会监视多路复用器负责的所有 socket，当任何一个 socket 的数据准备好了，多路复用器就会返回。这时候用户进程再调用 read 操作，把数据从内核缓冲区拷贝到用户空间。

　　所以，I/O 多路复用的特点是通过一种机制一个进程能同时等待多个文件描述符，而这些文件描述符（套接字描述符）其中的任意一个进入读就绪（readable）状态，select()函数就可以返回。Redis 的多路复用，提供了 select, epoll, evport, kqueue 几种选择，在编译的时候来选择一种。

evport 是 Solaris 系统内核提供支持的；
epoll 是 LINUX 系统内核提供支持的；
kqueue 是 Mac 系统提供支持的；
select 是 POSIX 提供的，一般的操作系统都有支撑（保底方案）；

　　我们看到 NIOClientSocket中下面这段代码，当客户端通过 read 方法去读取服务端返回的数据时，如果此时服务端数据未准备好，对于客户端来说就是一次无效的轮询。

while(true) {
    int i = socketChannel.read(byteBuffer);
    if (i > 0) {
        System.out.println("收到服务端的数据：" + new String(byteBuffer.array()));
    } else {
        System.out.println("服务端数据未准备好");
        Thread.sleep(1000);
    }
}

　　我们能不能够设计成，当客户端调用 read 方法之后，不仅仅不阻塞，同时也不需要轮询。而是等到服务端的数据就绪之后，告诉客户端。然后客户端再去读取服务端返回的数据呢？就像点外卖一样，我们在网上下单之后，继续做其他事情，等到外卖到了公司，外卖小哥主动打电话告诉你，你直接去前台取餐即可。

　　所以为了优化这个问题，引入了多路复用机制。

　　I/O多路复用的本质是通过一种机制（系统内核缓冲I/O数据），让单个进程可以监视多个文件描述符，一旦某个描述符就绪（一般是读就绪或写就绪），能够通知程序进行相应的读写操作

　　什么是fd：在linux中，内核把所有的外部设备都当成是一个文件来操作，对一个文件的读写会调用内核提供的系统命令，返回一个fd(文件描述符)。而对于一个socket的读写也会有相应的文件描述符，成为socketfd。

　　常见的IO多路复用方式有【select、poll、epoll】，都是Linux API提供的IO复用方式，那么接下来重点讲一下select、和epoll这两个模型

select：进程可以通过把一个或者多个fd传递给select系统调用，进程会阻塞在select操作上，这样select可以帮我们检测多个fd是否处于就绪状态，这个模式有两个缺点由于他能够同时监听多个文件描述符，假如说有1000个，这个时候如果其中一个fd 处于就绪状态了，那么当前进程需要线性轮询所有的fd，也就是监听的fd越多，性能开销越大。同时，select在单个进程中能打开的fd是有限制的，默认是1024，对于那些需要支持单机上万的TCP连接来说确实有点少
epoll：linux还提供了epoll的系统调用，epoll是基于事件驱动方式来代替顺序扫描，因此性能相对来说更高，主要原理是，当被监听的fd中，有fd就绪时，会告知当前进程具体哪一个fd就绪，那么当前进程只需要去从指定的fd上读取数据即可，另外，epoll所能支持的fd上线是操作系统的最大文件句柄，这个数字要远远大于1024

　　由于epoll能够通过事件告知应用进程哪个fd是可读的，所以我们也称这种IO为异步非阻塞IO，当然它是伪异步的，因为它还需要去把数据从内核同步复制到用户空间中，真正的异步非阻塞，应该是数据已经完全准备好了，我只需要从用户空间读就行.

　　同步和异步，指的是用户线程和内核的交互方式，阻塞和非阻塞，指用户线程调用内核IO操作的方式是阻塞还是非阻塞就像在Java中使用多线程做异步处理的概念，通过多线程去执行一个流程，主线程可以不用等待。而阻塞和非阻塞我们可以理解为假如在同步流程或者异步流程中做IO操作，如果缓冲区数据还没准备好，IO的这个过程会阻塞。

　　I/O多路复用的好处是可以通过把多个I/O的阻塞复用到同一个select的阻塞上，从而使得系统在单线程的情况下可以同时处理多个客户端请求。它的最大优势是系统开销小，并且不需要创建新的进程或者线程，降低了系统的资源开销，它的整体实现思想如下图所示。

　　客户端请求到服务端后，此时客户端在传输数据过程中，为了避免Server端在read客户端数据过程中阻塞，服务端会把该请求注册到Selector复路器上，服务端此时不需要等待，只需要启动一个线程，通过 selector.select()阻塞轮询复路器上就绪的channel即可。

　　也就是说，如果某个客户端连接数据传输完成，那么select()方法会返回就绪的channel，然后执行相关的处理即可。

　　　　代码如下：

public class NIOSelectorServerSocket implements Runnable{

    Selector selector;
    ServerSocketChannel serverSocketChannel;

    public NIOSelectorServerSocket(int port) throws IOException {
        selector=Selector.open();
        serverSocketChannel=ServerSocketChannel.open();
        //如果采用selector模型，必须要设置非阻塞
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    }

    @Override
    public void run() {
        while(!Thread.interrupted()){
            try {
                selector.select(); //阻塞等待事件就绪
                Set selected=selector.selectedKeys(); //事件列表
                Iterator it=selected.iterator();
                while(it.hasNext()){
                    //说明有连接进来
                    dispatch((SelectionKey) it.next());
                    it.remove();//移除当前就绪的事件
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    private void dispatch(SelectionKey key) throws IOException {
        if(key.isAcceptable()){ //是连接事件？
            register(key);
        }else if(key.isReadable()){ //读事件
            read(key);
        }else if(key.isWritable()){ //写事件
            //TODO
        }
    }
    private void register(SelectionKey key) throws IOException {
        ServerSocketChannel channel= (ServerSocketChannel) key.channel(); //客户端连接
        SocketChannel socketChannel=channel.accept(); //获得客户端连接
        socketChannel.configureBlocking(false);
        socketChannel.register(selector,SelectionKey.OP_READ);
    }
    private void read(SelectionKey key) throws IOException {
        //得到的是socketChannel
        SocketChannel channel= (SocketChannel) key.channel();
        ByteBuffer byteBuffer=ByteBuffer.allocate(1024);
        channel.read(byteBuffer);
        System.out.println("Server Receive Msg:"+new String(byteBuffer.array()));
    }

    public static void main(String[] args) throws IOException {
        NIOSelectorServerSocket selectorServerSocket=new NIOSelectorServerSocket(8080);
        new Thread(selectorServerSocket).start();
    }
}

　　事实上NIO已经解决了上述BIO暴露的下面两个问题：

同步阻塞IO，读写阻塞，线程等待时间过长。
在制定线程策略的时候，只能根据CPU的数目来限定可用线程资源，不能根据连接并发数目来制定，也就是连接有限制。否则很难保证对客户端请求的高效和公平。

　　到这里为止，通过NIO的多路复用机制，解决了IO阻塞导致客户端连接处理受限的问题，服务端只需要一个线程就可以维护多个客户端，并且客户端的某个连接如果准备就绪时，会通过事件机制告诉应用程序某个channel可用，应用程序通过select方法选出就绪的channel进行处理。

单线程Reactor 模型（高性能I/O设计模式）:

　　了解了NIO多路复用后，就有必要再和大家说一下Reactor多路复用高性能I/O设计模式，Reactor本质上就是基于NIO多路复用机制提出的一个高性能IO设计模式。

　　它的核心思想是把响应IO事件和业务处理进行分离，通过一个或者多个线程来处理IO事件，然后将就绪得到事件分发到业务处理handlers线程去步非阻塞处理，如下图所示。 Reactor模型有三个重要的组件：

Reactor ：将I/O事件发派给对应的Handler
Acceptor ：处理客户端连接请求
Handlers ：执行非阻塞读/写

　　代码实现如下：　　

　　Reactor：

public class Reactor implements Runnable {

    private final Selector selector;
    private final ServerSocketChannel serverSocketChannel;

    public Reactor(int port) throws IOException {
        selector = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, new Acceptor(selector, serverSocketChannel));
    }

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    dispatch(iterator.next());
                    iterator.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private void dispatch(SelectionKey key) {
        //可能拿到的对象有两个
        // Acceptor
        // Handler
        Runnable runnable = (Runnable) key.attachment();
        if (runnable != null) {
            runnable.run(); //
        }
    }
}

　　Acceptor：

public class Acceptor implements Runnable {

    private final Selector selector;
    private final ServerSocketChannel serverSocketChannel;

    public Acceptor(Selector selector, ServerSocketChannel serverSocketChannel) {
        this.selector = selector;
        this.serverSocketChannel = serverSocketChannel;
    }

    @Override
    public void run() {
        SocketChannel channel;

        try {
            channel = serverSocketChannel.accept();//得到一个客户端连接
            System.out.println(channel.getRemoteAddress() + ":收到一个客户端连接");
            channel.configureBlocking(false);
            channel.register(selector, SelectionKey.OP_READ, new Handler(channel));
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}

　　Handler ：

public class Handler implements Runnable {
    SocketChannel channel;

    public Handler(SocketChannel channe) {
        this.channel = channe;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + "------");
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        int len = 0, total = 0;
        String msg = "";
        try {
            do {
                len = channel.read(buffer);
                if (len > 0) {
                    total += len;
                    msg += new String(buffer.array());
                }
            } while (len > buffer.capacity());
            System.out.println("total:" + total);

            //msg=表示通信传输报文
            //耗时2s
            //登录： username:password
            //ServetRequets: 请求信息
            //数据库的判断
            //返回数据，通过channel写回到客户端

            System.out.println(channel.getRemoteAddress() + ": Server receive Msg:" + msg);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

　　以上代码码是最基本的单Reactor单线程模型（整体的I/O操作是由同一个线程完成的）。

　　其中Reactor线程，负责多路分离套接字，有新连接到来触发connect 事件之后，交由Acceptor进行处理，有IO读写事件之后交给hanlder 处理。

　　Acceptor主要任务就是构建handler ，在获取到和client相关的SocketChannel之后，绑定到相应的 hanlder上，对应的SocketChannel有读写事件之后，基于racotor 分发,hanlder就可以处理了（所有的 IO事件都绑定到selector上，有Reactor分发）。

　　Reactor 模式本质上指的是使用 I/O 多路复用(I/O multiplexing) + 非阻塞 I/O(nonblocking I/O) 的模式。

多线程单Reactor模型：

　　单线程Reactor这种实现方式有存在着缺点，从实例代码中可以看出，handler的执行是串行的，如果其中一个handler处理线程阻塞将导致其他的业务处理阻塞。由于handler和reactor在同一个线程中的执行，这也将导致新的无法接收新的请求，我们做一个小实验：

在上述Reactor代码的DispatchHandler的run方法中，增加一个Thread.sleep()。
打开多个客户端窗口连接（可以通过telnet连接）到Reactor Server端，其中一个窗口发送一个信息后被阻塞，另外一个窗口再发信息时由于前面的请求阻塞导致后续请求无法被处理。

　　Redis6.0中多线程带来的性能提升。Redis中的特殊的多线程单Reactor模型。下图是美团技术团队使用阿里云服务器压测GET/SET命令在4个线程IO时性能上的对比结果，可以明显的看到，Redis 在使用多线程模式之后性能大幅提升，达到了一倍。

　　Redis Server 阿里云 Ubuntu 18.04 ， 8CPU 2.5GHZ，8G内存，主机型号： ecs.ic5.2xlarge Redis Benchmark client: 阿里云 Unbuntu 18.04 , 8CPU 2.5GHZ，8G内存，主机型号： ecs.ic5.2xlarge

　　为了解决这种问题，有人提出使用多线程的方式来处理业务，也就是在业务处理的地方加入线程池异步处理，将reactor和handler在不同的线程来执行，如下图所示。

　　改造代码代码如下：

public class MutilDispatchHandler implements Runnable {

    SocketChannel channel;

    private Executor executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    public MutilDispatchHandler(SocketChannel channel) {
        this.channel = channel;
    }

    @Override
    public void run() {
        processor();
    }

    private void processor() {
        executor.execute(new ReaderHandler(channel));
    }

    static class ReaderHandler implements Runnable {
        private SocketChannel channel;

        public ReaderHandler(SocketChannel channel) {
            this.channel = channel;
        }

        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + ":-----");
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            try {
                Thread.sleep(100000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            int len = 0, total = 0;
            String msg = "";
            try {
                do {
                    len = channel.read(buffer);
                    if (len > 0) {
                        total += len;
                        msg += new String(buffer.array());
                    }
                } while (len > buffer.capacity());
                System.out.println("total:" + total);

                //msg=表示通信传输报文
                //耗时2s
                //登录： username:password
                //ServetRequets: 请求信息
                //数据库的判断
                //返回数据，通过channel写回到客户端
                System.out.println(channel.getRemoteAddress() + ": Server receive Msg:" + msg);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (channel != null) {
                    try {
                        channel.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}

　　在多线程Reactor模型中，添加了一个工作者线程池，并将非I/O操作从Reactor线程中移出转交给工作者线程池来执行。这样能够提高Reactor线程的I/O响应，不至于因为一些耗时的业务逻辑而延迟对后面 I/O请求的处理。

主从多线程多Reactor模式：

　　mainReactor负责监听连接，accept连接给subReactor处理，为什么要单独分一个Reactor来处理监听呢？因为像TCP这样需要经过3次握手才能建立连接，这个建立连接的过程也是要耗时间和资源的，单独分一个Reactor来处理，可以提高性能。

　　代码实现如下：

　　Acceptor

public class Acceptor implements Runnable{

    final Selector sel;
    final ServerSocketChannel serverSocketChannel;
    //获取当前核心数
    private final int POOL_SIZE=Runtime.getRuntime().availableProcessors();
    //线程池
    private Executor subReactorExecutor= Executors.newFixedThreadPool(POOL_SIZE);
    //subReactors 个数
    private Reactor[] subReactors=new Reactor[POOL_SIZE];
    //任务分发轮询分发计数
    int handerNext=0;

    public Acceptor(Selector sel,int port) throws IOException {
        this.sel=sel;//打开链接
        this.serverSocketChannel=ServerSocketChannel.open();
        this.serverSocketChannel.socket().bind(new InetSocketAddress(port));
        this.serverSocketChannel.configureBlocking(false);
        this.serverSocketChannel.register(this.sel, SelectionKey.OP_ACCEPT,this);
        init();
        System.out.println("Main Reactor Acceptor: Listening on port:"+port);
    }
    private void init() throws IOException {
        //初始化subReactors
        for (int i = 0; i < subReactors.length; i++) {
            subReactors[i]=new Reactor();
            subReactorExecutor.execute(subReactors[i]);
        }
    }
    @Override
    public void run() {
        //负责处理连接事件和IO事件
        try {
            SocketChannel socketChannel=serverSocketChannel.accept(); //获取连接
            if(socketChannel!=null){
                socketChannel.write(ByteBuffer.wrap("Multiply Reactor Patterm\r\nreactor> ".getBytes()));
                System.out.println(Thread.currentThread().getName()+": Main-Reactor-Acceptor:"+socketChannel.getLocalAddress()+"连接");
                //轮询分发
                Reactor subReactor=subReactors[handerNext];
                //异步处理
                subReactor.register(new AsyncHandler(socketChannel));
                if(++handerNext==subReactors.length){
                    handerNext=0;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

　　Reactor：

public class Reactor implements Runnable{

    private final Selector selector;
    //任务队列
    private ConcurrentLinkedQueue<AsyncHandler> events=new ConcurrentLinkedQueue<>();

    public Reactor() throws IOException {
        //打开selector
        this.selector = Selector.open();
    }

    public Selector getSelector() {
        return selector;
    }

    @Override
    public void run() {
        while(!Thread.interrupted()){
            AsyncHandler handler;
            try {
                //阻塞。获取任务队列，为空的时候会阻塞
                while((handler=events.poll())!=null){
                    handler.getChannel().configureBlocking(false);
                    SelectionKey selectionKey=handler.getChannel().register(selector,SelectionKey.OP_READ);
                    selectionKey.attach(handler);
                    handler.setSk(selectionKey);
                }
                //获取到了任务队列，此刻进入轮询监听IO事件
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while(iterator.hasNext()){
                    SelectionKey key=iterator.next();
                    Runnable runnable=(Runnable) key.attachment(); //得到AsyncHandler实例
                    if(runnable!=null){
                        //调用AsyncHandler 处理任务
                        runnable.run();
                    }
                    iterator.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

        }
    }

    public void register(AsyncHandler handler){
        events.offer(handler); //有一个事件注册，添加任务队列
        selector.wakeup();//唤醒阻塞selector
    }
}

　　AsyncHandler：

public class AsyncHandler implements Runnable{

    private SocketChannel channel;
    private SelectionKey sk;
    
    StringBuilder stringBuilder=new StringBuilder();

    ByteBuffer inputBuffer=ByteBuffer.allocate(1024);
    ByteBuffer outputBuffer=ByteBuffer.allocate(1024);

    public AsyncHandler(SocketChannel channel) {
        this.channel = channel;
    }

    public SocketChannel getChannel() {
        return channel;
    }

    public SelectionKey getSk() {
        return sk;
    }

    public void setSk(SelectionKey sk) {
        this.sk = sk;
    }

    @Override
    public void run() {
        try {
            if (sk.isReadable()) {
                read();
            } else if (sk.isWritable()) {
                write();
            }
        }catch (Exception e){
        }
    }
    private void read() throws IOException {
        inputBuffer.clear();
        int n=channel.read(inputBuffer);
        if(inputBufferComplete(n)){
            System.out.println(Thread.currentThread().getName()+": Server端收到客户端的请求消息："+stringBuilder.toString());
            outputBuffer.put(stringBuilder.toString().getBytes(StandardCharsets.UTF_8));
            this.sk.interestOps(SelectionKey.OP_WRITE);
        }
    }
    private boolean inputBufferComplete(int bytes) throws EOFException {
        if(bytes>0){
            inputBuffer.flip();
            while(inputBuffer.hasRemaining()){
                byte ch=inputBuffer.get(); //得到输入的字符
                if(ch==3) { //表示Ctrl+c
                    throw new EOFException();
                }else if(ch=='\r'||ch=='\n'){
                    return true;
                }else {
                    stringBuilder.append((char)ch);
                }
            }
        }else if(bytes==1){
            throw new EOFException();
        }
        return false;
    }

    private void write() throws IOException {
        int write=-1;
        outputBuffer.flip();
        if(outputBuffer.hasRemaining()){
            write=channel.write(outputBuffer); //把收到的数据写回到客户端
        }
        outputBuffer.clear();
        stringBuilder.delete(0,stringBuilder.length());
        if(write<=0){
            this.sk.channel().close();
        }else{
            channel.write(ByteBuffer.wrap("\r\nreactor> ".getBytes()));
            this.sk.interestOps(SelectionKey.OP_READ);//又转化为读事件
        }
    }
}

　　MultiplyReactor：

public class MultiplyReactor {

    private int port;

    private Reactor mainReactor; //main Reactor

    Executor mainReactorExecutor= Executors.newFixedThreadPool(10);

    public MultiplyReactor(int port) throws IOException {
        this.port = port;
        mainReactor=new Reactor();
    }

    public void start() throws IOException {
        new Acceptor(mainReactor.getSelector(),port);
        mainReactorExecutor.execute(mainReactor);
    }

    public static void main(String[] args) throws IOException {
        new MultiplyReactor(8080).start();
    }
}

　　父线程与子线程的数据交互简单职责明确，父线程只需要接收新连接，子线程完成后续的业务处理。父线程与子线程的数据交互简单，Reactor 主线程只需要把新连接传给子线程，子线程无需返回数据。

　　这种模型在许多项目中广泛使用，包括 Nginx 主从 Reactor 多进程模型，Memcached 主从多线程，Netty 主从多线程模型的支持。

标签：Reactor,redis,线程,内核,IO,磁盘,public,channel,客户端
From： https://www.cnblogs.com/cdlyy/p/17735602.html

redis为什么这么快、底层磁盘以及IO模型

Redis是单进程单线程？为什么这么快：

单线程为什么这么快？

从请求处理开始分析：

网络IO的通信原理：

BIO 传统阻塞IO ：

NIO非阻塞IO：

NIO多路复用机制：

单线程Reactor 模型（高性能I/O设计模式）:

多线程单Reactor模型：

相关文章

赞助商

阅读排行

redis为什么这么快、底层磁盘以及IO模型

Redis是单进程单线程？为什么这么快：

单线程为什么这么快？

从请求处理开始分析：

网络IO的通信原理 ：

BIO 传统阻塞IO ：

NIO非阻塞IO：

NIO多路复用机制：

单线程Reactor 模型（高性能I/O设计模式）:

多线程单Reactor模型：

相关文章

赞助商

阅读排行

网络IO的通信原理：