您现在的位置是：网站首页> 编程资料编程资料

Redis线程模型的原理分析_Redis_

2023-05-27 601人已围观

简介 Redis线程模型的原理分析_Redis_

一、概述

众所周知，Redis是一个高性能的数据存储框架，在高并发的系统设计中，Redis也是一个比较关键的组件，是我们提升系统性能的一大利器。深入去理解Redis高性能的原理显得越发重要，当然Redis的高性能设计是一个系统性的工程，涉及到很多内容，本文重点关注Redis的IO模型，以及基于IO模型的线程模型。

我们从IO的起源开始，讲述了阻塞IO、非阻塞IO、多路复用IO。基于多路复用IO，我们也梳理了几种不同的Reactor模型，并分析了几种Reactor模型的优缺点。基于Reactor模型我们开始了Redis的IO模型和线程模型的分析，并总结出Redis线程模型的优点、缺点，以及后续的Redis多线程模型方案。本文的重点是对Redis线程模型设计思想的梳理，捋顺了设计思想，就是一通百通的事了。

注：本文的代码都是伪代码，主要是为了示意，不可用于生产环境。

二、网络IO模型发展史

我们常说的网络IO模型，主要包含阻塞IO、非阻塞IO、多路复用IO、信号驱动IO、异步IO，本文重点关注跟Redis相关的内容，所以我们重点分析阻塞IO、非阻塞IO、多路复用IO，帮助大家后续更好的理解Redis网络模型。

我们先看下面这张图；

2.1 阻塞IO

我们经常说的阻塞IO其实分为两种，一种是单线程阻塞，一种是多线程阻塞。这里面其实有两个概念，阻塞和线程。

阻塞：指调用结果返回之前，当前线程会被挂起，调用线程只有在得到结果之后才会返回；
线程：系统调用的线程个数。

像建立连接、读、写都涉及到系统调用，本身是一个阻塞的操作。

2.1.1 单线程阻塞

服务端单线程来处理，当客户端请求来临时，服务端用主线程来处理连接、读取、写入等操作。

以下用代码模拟了单线程的阻塞模式；

 import java.net.Socket; public class BioTest { public static void main(String[] args) throws IOException { ServerSocket server=new ServerSocket(8081); while(true) { Socket socket=server.accept(); System.out.println("accept port:"+socket.getPort()); BufferedReader in=new BufferedReader(new InputStreamReader(socket.getInputStream())); String inData=null; try { while ((inData = in.readLine()) != null) { System.out.println("client port:"+socket.getPort()); System.out.println("input data:"+inData); if("close".equals(inData)) { socket.close(); } } } catch (IOException e) { e.printStackTrace(); } finally { try { socket.close(); } catch (IOException e) { e.printStackTrace(); } } } } }

我们准备用两个客户端同时发起连接请求、来模拟单线程阻塞模式的现象。同时发起连接，通过服务端日志，我们发现此时服务端只接受了其中一个连接，主线程被阻塞在上一个连接的read方法上。

我们尝试关闭第一个连接，看第二个连接的情况，我们希望看到的现象是，主线程返回，新的客户端连接被接受。

从日志中发现，在第一个连接被关闭后，第二个连接的请求被处理了，也就是说第二个连接请求在排队，直到主线程被唤醒，才能接收下一个请求，符合我们的预期。

此时不仅要问，为什么呢？

主要原因在于accept、read、write三个函数都是阻塞的，主线程在系统调用的时候，线程是被阻塞的，其他客户端的连接无法被响应。

通过以上流程，我们很容易发现这个过程的缺陷，服务器每次只能处理一个连接请求，CPU没有得到充分利用，性能比较低。如何充分利用CPU的多核特性呢？自然而然的想到了——多线程逻辑。

2.1.2 多线程阻塞

对工程师而言，代码解释一切，直接上代码。

BIO多线程

 package net.io.bio; import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import java.net.ServerSocket; import java.net.Socket; public class BioTest { public static void main(String[] args) throws IOException { final ServerSocket server=new ServerSocket(8081); while(true) { new Thread(new Runnable() { public void run() { Socket socket=null; try { socket = server.accept(); System.out.println("accept port:"+socket.getPort()); BufferedReader in=new BufferedReader(new InputStreamReader(socket.getInputStream())); String inData=null; while ((inData = in.readLine()) != null) { System.out.println("client port:"+socket.getPort()); System.out.println("input data:"+inData); if("close".equals(inData)) { socket.close(); } } } catch (IOException e) { e.printStackTrace(); } finally { } } }).start(); } } }

同样，我们并行发起两个请求；

两个请求，都被接受，服务端新增两个线程来处理客户端的连接和后续请求。

我们用多线程解决了，服务器同时只能处理一个请求的问题，但同时又带来了一个问题，如果客户端连接比较多时，服务端会创建大量的线程来处理请求，但线程本身是比较耗资源的，创建、上下文切换都比较耗资源，又如何去解决呢？

2.2 非阻塞

如果我们把所有的Socket（文件句柄，后续用Socket来代替fd的概念，尽量减少概念，减轻阅读负担）都放到队列里，只用一个线程来轮训所有的Socket的状态，如果准备好了就把它拿出来，是不是就减少了服务端的线程数呢？

一起看下代码，单纯非阻塞模式，我们基本上不用，为了演示逻辑，我们模拟了相关代码如下；

 package net.io.bio; import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import java.net.ServerSocket; import java.net.Socket; import java.net.SocketTimeoutException; import java.util.ArrayList; import java.util.List; import org.apache.commons.collections4.CollectionUtils; public class NioTest { public static void main(String[] args) throws IOException { final ServerSocket server=new ServerSocket(8082); server.setSoTimeout(1000); List sockets=new ArrayList(); while (true) { Socket socket = null; try { socket = server.accept(); socket.setSoTimeout(500); sockets.add(socket); System.out.println("accept client port:"+socket.getPort()); } catch (SocketTimeoutException e) { System.out.println("accept timeout"); } //模拟非阻塞：轮询已连接的socket，每个socket等待10MS，有数据就处理，无数据就返回，继续轮询 if(CollectionUtils.isNotEmpty(sockets)) { for(Socket socketTemp:sockets ) { try { BufferedReader in=new BufferedReader(new InputStreamReader(socketTemp.getInputStream())); String inData=null; while ((inData = in.readLine()) != null) { System.out.println("input data client port:"+socketTemp.getPort()); System.out.println("input data client port:"+socketTemp.getPort() +"data:"+inData); if("close".equals(inData)) { socketTemp.close(); } } } catch (SocketTimeoutException e) { System.out.println("input client loop"+socketTemp.getPort()); } } } } } }

系统初始化，等待连接；

发起两个客户端连接，线程开始轮询两个连接中是否有数据。

两个连接分别输入数据后，轮询线程发现有数据准备好了，开始相关的逻辑处理（单线程、多线程都可）。

再用一张流程图辅助解释下（系统实际采用文件句柄，此时用Socket来代替，方便大家理解）。

服务端专门有一个线程来负责轮询所有的Socket，来确认操作系统是否完成了相关事件，如果有则返回处理，如果无继续轮询，大家一起来思考下？此时又带来了什么问题呢。

CPU的空转、系统调用（每次轮询到涉及到一次系统调用，通过内核命令来确认数据是否准备好），造成资源的浪费，那有没有一种机制，来解决这个问题呢？

2.3 IO多路复用

server端有没专门的线程来做轮询操作（应用程序端非内核），而是由事件来触发，当有相关读、写、连接事件到来时，主动唤起服务端线程来进行相关逻辑处理。模拟了相关代码如下；

IO多路复用

 import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.nio.charset.Charset; import java.util.Iterator; import java.util.Set; public class NioServer { private static Charset charset = Charset.forName("UTF-8"); public static void main(String[] args) { try { Selector selector = Selector.open(); ServerSocketChannel chanel = ServerSocketChannel.open(); chanel.bind(new InetSocketAddress(8083)); chanel.configureBlocking(false); chanel.register(selector, SelectionKey.OP_ACCEPT); while (true){ int select = selector.select(); if(select == 0){ System.out.println("select loop"); continue; } System.out.println("os data ok"); Set selectionKeys = selector.selectedKeys(); Iterator iterator = selectionKeys.iterator(); while (iterator.hasNext()){ SelectionKey selectionKey = iterator.next(); if(selectionKey.isAcceptable()){ ServerSocketChannel server = (ServerSocketChannel)selectionKey.channel(); SocketChannel client = server.accept(); client.configureBlocking(false); client.register(selector, SelectionKey.OP_READ); //继续可以接收连接事件 selectionKey.interestOps(SelectionKey.OP_ACCEPT); }else if(selectionKey.isReadable()){ //得到SocketChannel SocketChannel client = (SocketChannel)selectionKey.channel(); //定义缓冲区 ByteBuffer buffer = ByteBuffer.allocate(1024); StringBuilder content = new StringBuilder(); while (client.read(buffer) > 0){ buffer.flip(); content.append(charset.decode(buffer)); } System.out.println("client port:"+client.getRemoteAddress().toString()+",input data: "+content.toString()); //清空缓冲区 buffer.clear(); } iterator.remove(); } } } catch (Exception e) { e.printStackTrace(); } } }

同时创建两个连接；