Redis的高性能网络IO模型

素颜马尾好姑娘i 2022-09-03 00:29 218阅读 0赞

Redis的性能由哪些因素决定？

内存：通过redis中间件提供的一些不同的类型来操作数据，数据实际上存放在内存中

CPU：CPU可以支持多线程

网络通信：需要基于网络通信去访问redis的数据结构去进行相关操作

## 网络IO的通信原理 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70][]

## 网络通信模型 ##

所有网络通信优化的本质都是增加客户端访问的连接数量

TCP/IP：通过IP:port访问目标服务的指定进程

### BIO（阻塞IO） ###

accept连接阻塞和IO阻塞，所以一旦出现网络或性能不高的情况，后面的客户端连接都会阻塞，直到前面的客户端释放连接

![2021073021033769.png][]

public class BIOServerSocket {
        public static void main(String[] args) {
            ServerSocket serverSocket = null;
            try {
                serverSocket = new ServerSocket(8080);
                while (true) {
                    Socket socket = serverSocket.accept(); //accept等待连接是阻塞的
                    System.out.println("客户端连接：" + socket.getInetAddress() + ":" + socket.getPort());
                    // I/O也是阻塞的
                    BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                    String clientMsg = bufferedReader.readLine();
                    System.out.println("收到客户端发送的消息：" + clientMsg);
                    BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
                    bufferedWriter.write("received message:" + clientMsg + "\n");
                    bufferedWriter.flush();
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (serverSocket != null) {
                        serverSocket.close();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

### 阻塞IO优化（IO不阻塞） ###

上面的阻塞IO，一旦由于网路问题或者电脑性能问题导致前面的客户端请求速度慢，那么后面的所有客户端请求都会被阻塞；所以引入了非阻塞IO，通过线程池创建多个线程，把IO处理部分交给每个线程去处理，从而将IO处理转为异步操作。在连接数要求不是很高的情况下可以使用这种模型。

**使用场景**：zookeeper的leader选举、nacos的注册地址信息同步

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 1][]

public class BIOServerSocketWithThreadPool {
        private static ExecutorService executorService = Executors.newFixedThreadPool(10);
        public static void main(String[] args) {
            ServerSocket serverSocket = null;
            try {
                serverSocket = new ServerSocket(8080);
                while (true) {
                    Socket socket = serverSocket.accept(); //accept等待连接是阻塞的
                    System.out.println("客户端连接：" + socket.getInetAddress() + ":" + socket.getPort());
                    // 将I/O阻塞部分扔给线程池创建的线程去处理，从而I/O变成了异步处理
                    executorService.submit(new ThreadSocket(socket));
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (serverSocket != null) {
                        serverSocket.close();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    
    public class ThreadSocket implements Runnable {
        private Socket socket = null;
    
        public ThreadSocket(Socket socket) {
            this.socket = socket;
        }
    
        @Override
        public void run() {
            try {
                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                String clientMsg = bufferedReader.readLine();
                System.out.println("收到客户端发送的消息：" + clientMsg);
                BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
                bufferedWriter.write("received message:" + clientMsg + "\n");
                bufferedWriter.flush();
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                // TODO 关闭I/O流避免占用资源
            }
        }
    }

### NIO ###

上面的非阻塞IO虽然解决了阻塞问题，但是线程池能够创建的线程数量是有限的，受限于CPU的线程数和核心数，不管多好的服务器，都无法满足我们想要达到的连接数，所以引入了NIO，把连接阻塞和IO阻塞改成非阻塞

NIO中核心特性：channel、buffer、selector

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 2][]

public class NIOServerSocket {
        public static void main(String[] args) {
            try {
                ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
                serverSocketChannel.configureBlocking(false); // 设置连接非阻塞
                while (true) {
                    // 获得客户端连接
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false); //设置IO非阻塞
                    if (null != socketChannel) {
                        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                        socketChannel.read(byteBuffer);
                        System.out.println(byteBuffer.array().toString());
                        byteBuffer.flip(); //反转
                        socketChannel.write(byteBuffer);
                    } else {
                        System.out.println("连接未就绪");
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

### NIO多路复用 ###

上面的NIO虽然解决了连接阻塞和IO阻塞，但是有一个弊端，当数据端数据在内核缓冲区没有就绪时，需要浪费CPU资源不停的轮询，而NIO多路复用模型正好解决这一弊端；多路复用模型的本质是通过事件机制，在内核层面监听多个fd，当数据可以读写时发送通知，通知客户端去连接，事件机制有3种：

Select模型：一个线程负责将所有的fd指令注册到select系统中，并轮询监听所有的fd指令，当数据准备就绪后再处理，单线程最多可打开fd数量是1024

Poll模型

Epoll模型

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 3][]

所有的客户端请求都会注册到selector来处理，一个线程轮询监听这些事件，一旦事件准备就绪，就会执行执行channel的任务。多路复用，多路指的是多个channel，复用指的是一个线程来轮询监听所有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.bind(new InetSocketAddress(port));
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        }
    
        // 定义个线程负责轮询selector
        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    selector.select(); // 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) throws IOException {
            // 判断事件类型，根据不同类型做不同处理
            if (key.isAcceptable()) { //连接事件
                this.register(key);
            } else if (key.isReadable()) { //读事件
                this.read(key);
            } else if (key.isWritable()) { //写事件
                this.write(key);
            }
        }
    
        // I/O事件也是阻塞的，所以也需要注册到多路复用器中
        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 channel = (SocketChannel) key.channel();
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            channel.read(byteBuffer);
            System.out.println("Server received message: " + new String(byteBuffer.array()));
        }
    
        private void write(SelectionKey key) throws IOException {
            SocketChannel channel = (SocketChannel) key.channel();
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            channel.write(byteBuffer);
        }
    
        public static void main(String[] args) throws IOException {
            NIOSelectorServerSocket selectorServerSocket = new NIOSelectorServerSocket(8080);
            new Thread(selectorServerSocket).start();
        }
    }

想了解更多网络通信模型知识请参照[https://blog.csdn.net/luciferlongxu/article/details/117167950][https_blog.csdn.net_luciferlongxu_article_details_117167950]

## Redis网络模型 ##

### 单线程Reactor模型 ###

基于NIO多路复用机制提出的高性能IO设计模式，把响应事件和业务进行分离，一个或多个线程处理IO事件，Redis6.0之前使用的此模型

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 4][]

Reactor：处理IO事件的分发

Handler：处理非阻塞事件的读和写

acceptor：处理客户端连接

public class Reactor implements Runnable {
        private Selector selector;
        private ServerSocketChannel serverSocketChannel;
    
        public Reactor(int port) throws IOException {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.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()) {
                        this.dispatch(iterator.next());
                        iterator.remove();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    
        private void dispatch(SelectionKey key) {
            // 此时收到的分发事件可能是acceptor，也可能是handler
            Runnable attachment = (Runnable) key.attachment();
            if (attachment != null) {
                attachment.run();
            }
        }
    }

public class Acceptor implements Runnable {
        private Selector selector;
        private ServerSocketChannel serverSocketChannel;
    
        public Acceptor(Selector selector, ServerSocketChannel serverSocketChannel) {
            this.selector = selector;
            this.serverSocketChannel = serverSocketChannel;
        }
    
        @Override
        public void run() {
            try {
                SocketChannel 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();
            }
        }
    }

public class Handler implements Runnable {
        private SocketChannel socketChannel;
    
        public Handler(SocketChannel socketChannel) {
            this.socketChannel = socketChannel;
        }
    
        @Override
        public void run() {
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            int length = 0;
            StringBuffer sb = new StringBuffer();
            try {
                do {
                    length = socketChannel.read(byteBuffer);
                    sb.append(new String(byteBuffer.array()));
                } while (length > byteBuffer.capacity());
                System.out.println(socketChannel.getRemoteAddress() + "接收到客户端消息:" + sb.toString());
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                if (null != socketChannel) {
                    try {
                        socketChannel.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

public class ReactoryTest {
        public static void main(String[] args) throws IOException {
            new Thread(new Reactor(8080)).start();
        }
    }

### 多线程Reactor模型 ###

单线程Reactor模型对事件的处理Handler是串行的，一旦一个事件的IO处理阻塞，后面的所有事件处理都需要阻塞等待；所以Redis6.0之后引入多线程Reactor模型，多线程Reactor模型通过线程池实现Handler是异步的

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 5][]

Reactor和Acceptor和前面一样，只有Handler事件处理使用了线程池多线程异步处理：

public class MultiThreadHandler implements Runnable {
        private SocketChannel socketChannel;
        private ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    
        public MultiThreadHandler(SocketChannel socketChannel) {
            this.socketChannel = socketChannel;
        }
    
        @Override
        public void run() {
            this.process();
        }
    
        private void process() {
            executorService.submit(new Processor(socketChannel));
        }
    }

public class Processor implements Runnable {
        private SocketChannel socketChannel;
        public Processor(SocketChannel socketChannel) {
            this.socketChannel = socketChannel;
        }
        @Override
        public void run() {
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            int length = 0;
            StringBuffer stringBuffer = new StringBuffer();
            try {
                do {
                    length = socketChannel.read(byteBuffer);
                    stringBuffer.append(new String(byteBuffer.array()));
                } while (length > byteBuffer.capacity());
                System.out.println(socketChannel.getRemoteAddress() + "收到客户端消息：" + stringBuffer.toString());
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

### 多线程多Reactor模型 ###

MainReactor负责接收客户端连接，接收到后把连接分发给不同的SubReactor去处理，SubReactor负责IO读写事件，并将IO事件交给Handler去处理

public class MultiReactor {
        private int port;
        private Reactor mainReactor;
        private ExecutorService executorService = Executors.newFixedThreadPool(10);
    
        public MultiReactor(int port) throws IOException {
            this.port = port;
            this.mainReactor = new Reactor();
        }
    
        public void start() throws IOException {
            new Acceptor(mainReactor.getSelector(), port);
            executorService.submit(mainReactor);
        }
    
        public static void main(String[] args) throws IOException {
            new MultiReactor(8080).start();
        }
    }

public class Reactor implements Runnable {
        private Selector selector;
        private ConcurrentLinkedQueue<Handler> queue = new ConcurrentLinkedQueue();
    
        public Selector getSelector() {
            return selector;
        }
    
        public Reactor() throws IOException {
            selector = Selector.open();
        }
    
        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    Handler handler = null;
                    while ((handler = queue.poll()) != null) { //可以使用阻塞队列
                        handler.getSocketChannel().configureBlocking(false);
                        SelectionKey selectionKey = handler.getSocketChannel().register(this.selector, SelectionKey.OP_READ, handler);
                        handler.setSelectionKey(selectionKey);
                    }
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey selectionKey = iterator.next();
                        Runnable attachment = (Runnable) selectionKey.attachment();//得到Acceptor实例
                        if (null != attachment) {
                            attachment.run();
                        }
                        iterator.remove();
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        public void register(Handler handler) {
            queue.offer(handler);
            this.selector.wakeup(); //唤醒selector
        }
    }

public class Acceptor implements Runnable {
        private Selector selector;
        private ServerSocketChannel serverSocketChannel;
        private static final int POOL_SIZE = Runtime.getRuntime().availableProcessors();
        private ExecutorService executorService = Executors.newFixedThreadPool(POOL_SIZE);
        private Reactor[] subReactors = new Reactor[POOL_SIZE];
        private int handlerIndex = 0;
    
        public Acceptor(Selector selector, int port) throws IOException {
            this.selector = selector;
            this.serverSocketChannel = ServerSocketChannel.open();
            this.serverSocketChannel.bind(new InetSocketAddress(port));
            this.serverSocketChannel.configureBlocking(false);
            this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT, this);
            this.init();
        }
    
        private void init() throws IOException {
            for (int i = 0; i < subReactors.length; i++) {
                subReactors[i] = new Reactor();
                executorService.submit(subReactors[i]);
            }
        }
    
        @Override
        public void run() {
            try {
                SocketChannel socketChannel = serverSocketChannel.accept(); //获取连接
                if (null != socketChannel) {
                    socketChannel.write(ByteBuffer.wrap("Multiple Reactor\r\nreactor> ".getBytes()));
                    System.out.println(Thread.currentThread().getName() + ":Main Reactor Acceptor:" + socketChannel.getLocalAddress() + "连接");
                    Reactor subReactor = subReactors[handlerIndex];
                    subReactor.register(new Handler(socketChannel));
                    if (++handlerIndex == subReactors.length) {
                        handlerIndex = 0;
                    }
                    //socketChannel.register(this.selector, SelectionKey.OP_READ);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

public class Handler implements Runnable {
        private SocketChannel socketChannel;
        private SelectionKey selectionKey;
        private ByteBuffer inputByteBuffer = ByteBuffer.allocate(1024);
        private ByteBuffer outputByteBuffer = ByteBuffer.allocate(1024);
        private StringBuffer stringBuffer = new StringBuffer();
    
        public Handler(SocketChannel socketChannel) {
            this.socketChannel = socketChannel;
        }
    
        public SocketChannel getSocketChannel() {
            return socketChannel;
        }
    
        public void setSelectionKey(SelectionKey selectionKey) {
            this.selectionKey = selectionKey;
        }
    
        public SelectionKey getSelectionKey() {
            return selectionKey;
        }
    
        @Override
        public void run() {
            try {
                if (selectionKey.isReadable()) {
                    this.read();
                } else if (selectionKey.isWritable()) {
                    this.write();
                }
            } catch (Exception e) {
    
            }
        }
    
        private void read() throws IOException {
            inputByteBuffer.clear();
            int read = socketChannel.read(inputByteBuffer);
            if (inputBufferComplete(read)) {
                System.out.println(Thread.currentThread().getName()+"Server端收到客户端的消息：" + stringBuffer.toString());
                this.outputByteBuffer.put(stringBuffer.toString().getBytes(StandardCharsets.UTF_8));
                this.selectionKey.interestOps(SelectionKey.OP_WRITE);//转换事件类型为写
            }
        }
    
        private boolean inputBufferComplete(int bytes) throws EOFException {
            if (bytes > 0) {
                inputByteBuffer.flip();
                while (inputByteBuffer.hasRemaining()) {
                    byte b = inputByteBuffer.get();//得到输入的字符
                    if (b == 3) {//ctrl + c
                        throw new EOFException();
                    } else if (b == '\r' || b == '\n') {
                        return true;
                    } else {
                        stringBuffer.append((char) b);
                    }
                }
            } else if (bytes == 1) {
                throw new EOFException();
            }
            return false;
        }
    
        private void write() throws IOException {
            int write = -1;
            outputByteBuffer.flip();
            if (outputByteBuffer.hasRemaining()) {
                write = this.socketChannel.write(outputByteBuffer);//把收到的数据写到客户端
            }
            outputByteBuffer.clear();
            stringBuffer.delete(0, stringBuffer.length());
            if (write < 0) {
                this.selectionKey.channel().close();
            } else {
                this.socketChannel.write(ByteBuffer.wrap("\r\nreactor> ".getBytes()));
                this.selectionKey.interestOps(SelectionKey.OP_READ); //又转化为读事件
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70]: /images/20220829/9f0804c8d7734bbe92ca0149bfed1725.png
[2021073021033769.png]: /images/20220829/d5f821ad6de346ccbadfdf3c592ba93d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 1]: /images/20220829/2d4718de63cb4e96b15e8e46ff11bef7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 2]: /images/20220829/bd6860605c0b48ba8664da38fbc2a2dd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 3]: /images/20220829/a49e68ba411b4993a172bf0e006c30b8.png
[https_blog.csdn.net_luciferlongxu_article_details_117167950]: https://blog.csdn.net/luciferlongxu/article/details/117167950
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 4]: /images/20220829/6bc85c45d0be40129af9bb868cc09f4b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 5]: /images/20220829/e14391740edf45e19608cb7a03058eb2.png