我們一直說Redis的性能很快，那為什么快？Redis為了達到性能最大化，做了哪些方面的優化呢？
在深度決議Redis的資料結構
這篇文章中，其實從資料結構上分析了Redis性能高的一方面原因，

在目前的k-v資料庫的技術選型中，Redis幾乎是首選的用來實作高性能快取的方案，它的性能有多快呢？

根據官方的基準測驗資料，一臺普通硬體配置的Linux機器上運行單個Redis實體，處理簡單命令（O(n)或者O（logn）），QPS可以達到8W，如果使用pipeline批處理功能，QPS最高可以達到10W，

Redis 為什么那么快

Redis的高性能主要依賴于幾個方面，

C語言實作，C語言在一定程度上還是比Java語言性能要高一些，因為C語言不需要經過JVM進行翻譯，
純記憶體I/O，記憶體I/O比磁盤I/O性能更快
I/O多路復用，基于epoll的I/O多路復用技術，實作高吞吐網路I/O
單執行緒模型，單執行緒無法利用到多核CPU，但是在Redis中，性能瓶頸并不是在計算上，而是在I/O能力，所以單執行緒能夠滿足高并發的要求，從另一個層面來說，單執行緒可以避免多執行緒的頻繁背景關系切換以及同步鎖機制帶來的性能開銷，

下面我們分別從上述幾個方面進行展開說明，先來看網路I/O的多路復用模型，

從請求處理開始分析

當我們在客戶端向Redis Server發送一條指令，并且得到Redis回復的整個程序中，Redis做了什么呢？

圖4-1

要處理命令，則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回傳時，資料不一定會被發送到對端，

1576066931883

網卡中的緩沖區既不屬于內核空間，也不屬于用戶空間，它屬于硬體緩沖，允許網卡與作業系統之間有個緩沖；
內核緩沖區在內核空間，在記憶體中，用于內核程式，做為讀自或寫往硬體的資料緩沖區；
用戶緩沖區在用戶空間，在記憶體中，用于用戶程式，做為讀自或寫往硬體的資料緩沖區

網卡芯片收到網路資料會以中斷的方式通知CPU，我有資料了，存在我的硬體緩沖里了，來讀我啊，
CPU收到這個中斷信號后，會呼叫相應的驅動介面函式從網卡的硬體緩沖里把資料讀到內核緩沖區，正常情況下會向上傳遞給TCP/IP模塊一層一層的處理，

NIO多路復用機制

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通信模型

BIOServerSocket

public class BIOServerSocket {
    //先定義一個埠號，這個埠的值是可以自己調整的，
    static final int DEFAULT_PORT=8080;
    public static void main(String[] args) throws IOException {
        //先定義一個埠號，這個埠的值是可以自己調整的，
        //在服務器端，我們需要使用ServerSocket，所以我們先宣告一個ServerSocket變數
        ServerSocket serverSocket=null;
        //接下來，我們需要系結監聽埠, 那我們怎么做呢？只需要創建使用serverSocket實體
        //ServerSocket有很多構造多載，在這里，我們把前邊定義的埠傳入，表示當前
        //ServerSocket監聽的埠是8080
        serverSocket=new ServerSocket(DEFAULT_PORT);
        System.out.println("啟動服務，監聽埠："+DEFAULT_PORT);
        //回顧一下前面我們講的內容，接下來我們就需要開始等待客戶端的連接了，
        //所以我們要使用的是accept這個函式，并且當accept方法獲得一個客戶端請求時，會回傳
        //一個socket物件， 這個socket物件讓服務器可以用來和客戶端通信的一個端點，

        //開始等待客戶端連接，如果沒有客戶端連接，就會一直阻塞在這個位置
        Socket socket=serverSocket.accept();
        //很可能有多個客戶端來發起連接，為了區分客戶端，咱們可以輸出客戶端的埠號
        System.out.println("客戶端："+socket.getPort()+"已連接");
        //一旦有客戶端連接過來，我們就可以用到IO來獲得客戶端傳過來的資料，
        //使用InputStream來獲得客戶端的輸入資料
        //bufferedReader大家還記得吧，他維護了一個緩沖區可以減少資料源讀取的頻率
        BufferedReader bufferedReader=new BufferedReader(new InputStreamReader(socket.getInputStream()));
        String clientStr=bufferedReader.readLine(); //讀取一行資訊
        System.out.println("客戶端發了一段訊息："+clientStr);
        //服務端收到資料以后，可以給到客戶端一個回復，這里咱們用到BufferedWriter
        BufferedWriter bufferedWriter=new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
        bufferedWriter.write("我已經收到你的訊息了\n");
        bufferedWriter.flush(); //清慷訓沖區觸發訊息發送
    }
}

BIOClientSocket

public class BIOClientSocket {
    static final int DEFAULT_PORT=8080;
    public static void main(String[] args) throws IOException {

        //在客戶端這邊，咱們使用socket來連接到指定的ip和埠
        Socket socket=new Socket("localhost",8080);
        //使用BufferedWriter，像服務器端寫入一個訊息
        BufferedWriter bufferedWriter=new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
        bufferedWriter.write("我是客戶端Client-01\n");
        bufferedWriter.flush();
        BufferedReader bufferedReader=new BufferedReader(new InputStreamReader(socket.getInputStream()));
        String serverStr=bufferedReader.readLine(); //通過bufferedReader讀取服務端回傳的訊息
        System.out.println("服務端回傳的訊息："+serverStr);
    }
}

上述代碼構建了一個簡單的BIO通信模型，也就是服務端建立一個監聽，客戶端向服務端發送一個訊息，實作簡單的網路通信，那BIO有什么弊端呢？

我們通過對BIOServerSocket進行改造，關注case1和case2部分，

case1: 增加了while回圈，實作重復監聽
case2: 當服務端收到客戶端的請求后，不直接回傳，而是等待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(); //清慷訓沖區觸發訊息發送
        }
    }
}

接著，把BIOClientSocket復制兩份（client1、client2），同時向BIOServerSocket發起請求，

運行后看到的現象應該是： client1先發送請求到Server端，由于Server端等待20s才回傳，導致client2的請求一直被阻塞，

這個情況會導致一個問題，如果服務端在同一個時刻只能處理一個客戶端的連接，而如果一個網站同時有1000個用戶訪問，那么剩下的999個用戶都需要等待，而這個等待的耗時取決于前面的請求的處理時長，如圖4-2所示，

圖4-2

基于多執行緒優化BIO

為了讓服務端能夠同時處理更多的客戶端連接，避免因為某個客戶端連接阻塞導致后續請求被阻塞，于是引入多執行緒技術，代碼如下，

ServerSocket

public static void main(String[] args) throws IOException, InterruptedException {
    final int DEFAULT_PORT=8080;
    ServerSocket serverSocket=null;
    serverSocket=new ServerSocket(DEFAULT_PORT);
    System.out.println("啟動服務，監聽埠："+DEFAULT_PORT);
    ExecutorService executorService= Executors.newFixedThreadPool(5);
    while(true) {
        Socket socket = serverSocket.accept();
        executorService.submit(new SocketThread(socket));
    }
}

SocketThread

public class SocketThread implements Runnable{
    Socket socket;

    public SocketThread(Socket socket) {
        this.socket = socket;
    }
    @Override
    public void run() {
        System.out.println("客戶端：" + socket.getPort() + "已連接");
        try {
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            String clientStr = null; //讀取一行資訊
            clientStr = bufferedReader.readLine();
            System.out.println("客戶端發了一段訊息：" + clientStr);
            Thread.sleep(20000);
            BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
            bufferedWriter.write("我已經收到你的訊息了\n");
            bufferedWriter.flush(); //清慷訓沖區觸發訊息發送
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

如圖4-3所示，當引入了多執行緒之后，每個客戶端的鏈接（Socket），我們可以直接給到執行緒池去執行，而由于這個程序是異步的，所以并不會同步阻塞影響后續鏈接的監聽，因此在一定程度上可以提升服務端鏈接的處理數量，

圖4-3

NIO非阻塞IO

使用多執行緒的方式來解決這個問題，仍然有一個缺點，執行緒的數量取決于硬體配置，所以執行緒數量是有限的，如果請求量比較大的時候，執行緒本身會收到限制從而并發量也不會太高，那怎么辦呢，我們可以采用非阻塞IO，

NIO 從JDK1.4 提出的，本意是New IO，它的出現為了彌補原本IO的不足，提供了更高效的方式，提出一個通道（channel）的概念，在IO中它始終以流的形式對資料的傳輸和接受，下面我們演示一下NIO的使用，

NioServerSocket

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();
                if (socketChannel != null) {
                    //讀取資料
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    socketChannel.read(buffer);
                    System.out.println(new String(buffer.array()));
                    //寫出資料
                    Thread.sleep(10000); //阻塞一段時間
                    //當資料讀取到緩沖區之后，接下來就需要把緩沖區的資料寫出到通道，而在寫出之前必須要呼叫flip方法，實際上就是重置一個有效位元組范圍，然后把這個資料接觸到通道，
                    buffer.flip();
                    socketChannel.write(buffer);//寫出資料
                } else {
                    Thread.sleep(1000);
                    System.out.println("連接未就緒");
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

NioClientSocket

public class NioClientSocket {
    public static void main(String[] args) {
        try {
            SocketChannel socketChannel= SocketChannel.open();
            socketChannel.configureBlocking(false);
            socketChannel.connect(new InetSocketAddress("localhost",8080));
            if(socketChannel.isConnectionPending()){
                socketChannel.finishConnect();
            }
            ByteBuffer byteBuffer= ByteBuffer.allocate(1024);
            byteBuffer.put("Hello I'M SocketChannel Client".getBytes());
            byteBuffer.flip();
            socketChannel.write(byteBuffer);
            //讀取服務端資料
            byteBuffer.clear();
            while(true) {
                int i = socketChannel.read(byteBuffer);
                if (i > 0) {
                    System.out.println("收到服務端的資料：" + new String(byteBuffer.array()));
                } else {
                    System.out.println("服務端資料未準備好");
                    Thread.sleep(1000);
                }
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }

    }
}

所謂的NIO（非阻塞IO），其實就是取消了IO阻塞和連接阻塞，當服務端不存在阻塞的時候，就可以不斷輪詢處理客戶端的請求，如圖4-4所示，表示NIO下的運行流程，

圖4-4

上述這種NIO的使用方式，仍然存在一個問題，就是客戶端或者服務端需要通過一個執行緒不斷輪詢才能獲得結果，而這個輪詢程序中會浪費執行緒資源，

多路復用IO

大家站在全域的角度再思考一下整個程序，有哪些地方可以優化呢？

我們回到NIOClientSocket中下面這段代碼，當客戶端通過read方法去讀取服務端回傳的資料時，如果此時服務端資料未準備好，對于客戶端來說就是一次無效的輪詢，

我們能不能夠設計成，當客戶端呼叫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);
    }
}

所以為了優化這個問題，引入了多路復用機制，

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，當然它是偽異步的，因為它還需要去把資料從內核同步復制到用戶空間中，真正的異步非阻塞，應該是資料已經完全準備好了，我只需要從用戶空間讀就行】

I/O多路復用的好處是可以通過把多個I/O的阻塞復用到同一個select的阻塞上，從而使得系統在單執行緒的情況下可以同時處理多個客戶端請求，它的最大優勢是系統開銷小，并且不需要創建新的行程或者執行緒，降低了系統的資源開銷，它的整體實作思想如圖4-5所示，

客戶端請求到服務端后，此時客戶端在傳輸資料程序中，為了避免Server端在read客戶端資料程序中阻塞，服務端會把該請求注冊到Selector復路器上，服務端此時不需要等待，只需要啟動一個執行緒，通過selector.select()阻塞輪詢復路器上就緒的channel即可，也就是說，如果某個客戶端連接資料傳輸完成，那么select()方法會回傳就緒的channel，然后執行相關的處理即可，

圖4-5

NIOServer的實作如下

測驗訪問的時候，直接在cmd中通過telnet連接NIOServer，便可發送資訊，

public class NIOServer implements Runnable{
    Selector selector;
    ServerSocketChannel serverSocketChannel;
    public NIOServer(int port) throws IOException {
        selector=Selector.open(); //多路復用器
        serverSocketChannel=ServerSocketChannel.open();
        //系結監聽埠
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        serverSocketChannel.configureBlocking(false);//非阻塞配置
        //針對serverSocketChannel注冊一個ACCEPT連接監聽事件
        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()){
            write(key);
        }
    }
    private void register(SelectionKey key) throws IOException {
        //得到事件對應的連接
        ServerSocketChannel server=(ServerSocketChannel)key.channel();
        SocketChannel channel=server.accept(); //獲得客戶端的鏈接
        channel.configureBlocking(false);
        //把當前客戶端連接注冊到selector上，注冊事件為READ，
        // 也就是當前channel可讀時，就會觸發事件，然后讀取客戶端的資料
        channel.register(this.selector,SelectionKey.OP_READ);
    }

    private void read(SelectionKey key) throws IOException {
        SocketChannel channel=(SocketChannel)key.channel();
        ByteBuffer byteBuffer= ByteBuffer.allocate(1024);
        channel.read(byteBuffer); //把資料從channel讀取到緩沖區
        System.out.println("server receive msg:"+new String(byteBuffer.array()));
    }
    private void write(SelectionKey key) throws IOException {
        SocketChannel channel=(SocketChannel)key.channel();
        //寫一個資訊給到客戶端
        channel.write(ByteBuffer.wrap("hello Client,I'm NIO Server\r\n".getBytes()));
    }

    public static void main(String[] args) throws IOException {
        NIOServer server=new NIOServer(8888);
        new Thread(server).start();
    }
}

事實上NIO已經解決了上述BIO暴露的下面兩個問題：

同步阻塞IO，讀寫阻塞，執行緒等待時間過長，
在制定執行緒策略的時候，只能根據CPU的數目來限定可用執行緒資源，不能根據連接并發數目來制定，也就是連接有限制，否則很難保證對客戶端請求的高效和公平，

到這里為止，通過NIO的多路復用機制，解決了IO阻塞導致客戶端連接處理受限的問題，服務端只需要一個執行緒就可以維護多個客戶端，并且客戶端的某個連接如果準備就緒時，會通過事件機制告訴應用程式某個channel可用，應用程式通過select方法選出就緒的channel進行處理，

單執行緒Reactor 模型（高性能I/O設計模式）

了解了NIO多路復用后，就有必要再和大家說一下Reactor多路復用高性能I/O設計模式，Reactor本質上就是基于NIO多路復用機制提出的一個高性能IO設計模式，它的核心思想是把回應IO事件和業務處理進行分離，通過一個或者多個執行緒來處理IO事件，然后將就緒得到事件分發到業務處理handlers執行緒去異步非阻塞處理，如圖4-6所示，

Reactor模型有三個重要的組件：

Reactor ：將I/O事件發派給對應的Handler
Acceptor ：處理客戶端連接請求
Handlers ：執行非阻塞讀/寫

圖4-6

下面演示一個單執行緒的Reactor模型，

Reactor

Reactor 負責回應IO事件，一旦發生，廣播發送給相應的Handler去處理，

public class Reactor implements Runnable{
    private final Selector selector;
    private final ServerSocketChannel serverSocketChannel;

    public Reactor(int port) throws IOException {
        //創建選擇器
        selector= Selector.open();
        //創建NIO-Server
        serverSocketChannel=ServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress(port));
        serverSocketChannel.configureBlocking(false);
        SelectionKey key=serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        // 系結一個附加物件
        key.attach(new Acceptor(selector,serverSocketChannel));
    }

    @Override
    public void run() {
        while(!Thread.interrupted()){
            try {
                selector.select(); //阻塞等待就緒事件
                Set selectionKeys=selector.selectedKeys();
                Iterator it=selectionKeys.iterator();
                while(it.hasNext()){
                    dispatch((SelectionKey) it.next());
                    it.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    public void dispatch(SelectionKey key){
        //呼叫之前注冊時附加的物件，也就是attach附加的acceptor
        Runnable r=(Runnable)key.attachment();
        if(r!=null){
            r.run();
        }
    }

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

Acceptor

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() {
        SocketChannel channel;
        try {
            channel=serverSocketChannel.accept();
            System.out.println(channel.getRemoteAddress()+": 收到一個客戶端連接");
            channel.configureBlocking(false);
            //當channel連接中資料就緒時，呼叫DispatchHandler來處理channel
            //巧妙使用了SocketChannel的attach功能，將Hanlder和可能會發生事件的channel鏈接在一起，當發生事件時，可以立即觸發相應鏈接的Handler，
            channel.register(selector, SelectionKey.OP_READ,new DispatchHandler(channel));
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}

Handler

public class DispatchHandler implements Runnable{
    private SocketChannel channel;

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

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName()+"---handler"); //case: 列印當前執行緒名稱，證明I/O是同一個執行緒來處理，
        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());
                }
                buffer.clear();
            } while (len > buffer.capacity());
            System.out.println(channel.getRemoteAddress()+":Server Receive msg:"+msg);

        }catch (Exception e){
            e.printStackTrace();
            if(channel!=null){
                try {
                    channel.close();
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
            }
        }

    }
}

演示方式，通過window的cmd視窗，使用telnet 192.168.1.102 8888 連接到Server端進行資料通信；也可以通過下面這樣一個客戶端程式來訪問，

ReactorClient

public class ReactorClient {

    private static Selector selector;
    public static void main(String[] args) throws IOException {
        selector=Selector.open();
        //創建一個連接通道連接指定的server
        SocketChannel socketChannel= SocketChannel.open();
        socketChannel.configureBlocking(false);
        socketChannel.connect(new InetSocketAddress("192.168.1.102",8888));
        socketChannel.register(selector, SelectionKey.OP_CONNECT);
        while(true){
            selector.select();
            Set<SelectionKey> selectionKeys=selector.selectedKeys();
            Iterator<SelectionKey> iterator=selectionKeys.iterator();
            while(iterator.hasNext()){
                SelectionKey key=iterator.next();
                iterator.remove();
                if(key.isConnectable()){
                    handleConnection(key);
                }else if(key.isReadable()){
                    handleRead(key);
                }
            }
        }
    }
    private static void handleConnection(SelectionKey key) throws IOException {
        SocketChannel socketChannel=(SocketChannel)key.channel();
        if(socketChannel.isConnectionPending()){
            socketChannel.finishConnect();
        }
        socketChannel.configureBlocking(false);
        while(true) {
            Scanner in = new Scanner(System.in);
            String msg = in.nextLine();
            socketChannel.write(ByteBuffer.wrap(msg.getBytes()));
            socketChannel.register(selector,SelectionKey.OP_READ);
        }
    }
    private static void handleRead(SelectionKey key) throws IOException {
        SocketChannel channel=(SocketChannel)key.channel();
        ByteBuffer byteBuffer=ByteBuffer.allocate(1024);
        channel.read(byteBuffer);
        System.out.println("client receive msg:"+new String(byteBuffer.array()));
    }
}

這是最基本的單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(non-blocking I/O)的模式，

多執行緒單Reactor模型

單執行緒Reactor這種實作方式有存在著缺點，從實體代碼中可以看出，handler的執行是串行的，如果其中一個handler處理執行緒阻塞將導致其他的業務處理阻塞，由于handler和reactor在同一個執行緒中的執行，這也將導致新的無法接收新的請求，我們做一個小實驗：

在上述Reactor代碼的DispatchHandler的run方法中，增加一個Thread.sleep()，
打開多個客戶端視窗連接到Reactor Server端，其中一個視窗發送一個資訊后被阻塞，另外一個視窗再發資訊時由于前面的請求阻塞導致后續請求無法被處理，

為了解決這種問題，有人提出使用多執行緒的方式來處理業務，也就是在業務處理的地方加入執行緒池異步處理，將reactor和handler在不同的執行緒來執行，如圖4-7所示，

圖4-7

多執行緒改造-MultiDispatchHandler

我們直接將4.2.5小節中的Reactor單執行緒模型改成多執行緒，其實我們就是把IO阻塞的問題通過異步的方式做了優化，代碼如下，

public class MultiDispatchHandler implements Runnable{
    private SocketChannel channel;

    public MultiDispatchHandler(SocketChannel channel) {
        this.channel = channel;
    }
    private static Executor executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() << 1);

    @Override
    public void run() {
        processor();
    }
    private void processor(){
        executor.execute(new ReaderHandler(channel));
    }
    public static class ReaderHandler implements Runnable{
        private SocketChannel channel;

        public ReaderHandler(SocketChannel socketChannel) {
            this.channel = socketChannel;
        }
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+"---handler"); //case: 列印當前執行緒名稱，證明I/O是同一個執行緒來處理，
            ByteBuffer buffer= ByteBuffer.allocate(1024);
            int len=0;
            String msg="";
            try {
                do {
                    len = channel.read(buffer);
                    if (len > 0) {
                        msg += new String(buffer.array());
                    }
                    buffer.clear();
                } while (len > buffer.capacity());

                if(len>0) {
                    System.out.println(channel.getRemoteAddress() + ":Server Receive msg:" + msg);
                }
            }catch (Exception e){
                e.printStackTrace();
                if(channel!=null){
                    try {
                        channel.close();
                    } catch (IOException ioException) {
                        ioException.printStackTrace();
                    }
                }
            }
        }
    }
}

Acceptor

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() {
        SocketChannel channel;
        try {
            channel=serverSocketChannel.accept();
            System.out.println(channel.getRemoteAddress()+": 收到一個客戶端連接");
            channel.configureBlocking(false);
            //當channel連接中資料就緒時，呼叫DispatchHandler來處理channel
            //巧妙使用了SocketChannel的attach功能，將Hanlder和可能會發生事件的channel鏈接在一起，當發生事件時，可以立即觸發相應鏈接的Handler，
            channel.register(selector, SelectionKey.OP_READ,new MultiDispatchHandler(channel));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

多執行緒Reactor總結

在多執行緒Reactor模型中，添加了一個作業者執行緒池，并將非I/O操作從Reactor執行緒中移出轉交給作業者執行緒池來執行，這樣能夠提高Reactor執行緒的I/O回應，不至于因為一些耗時的業務邏輯而延遲對后面I/O請求的處理，

多Reactor多執行緒模式（主從多Reactor模型）

在多執行緒單Reactor模型中，我們發現所有的I/O操作是由一個Reactor來完成，而Reactor運行在單個執行緒中，它需要處理包括Accept()/read()/write/connect操作，對于小容量的場景，影響不大，但是對于高負載、大并發或大資料量的應用場景時，容易成為瓶頸，主要原因如下：

一個NIO執行緒同時處理成百上千的鏈路，性能上無法支撐，即便NIO執行緒的CPU負荷達到100%，也無法滿足海量訊息的讀取和發送；
當NIO執行緒負載過重之后，處理速度將變慢，這會導致大量客戶端連接超時，超時之后往往會進行重發，這更加重了NIO執行緒的負載，最侄訓導致大量訊息積壓和處理超時，成為系統的性能瓶頸；

所以，我們還可以更進一步優化，引入多Reactor多執行緒模式，如圖4-8所示，Main Reactor負責接收客戶端的連接請求，然后把接收到的請求傳遞給SubReactor（其中subReactor可以有多個），具體的業務IO處理由SubReactor完成，

Multiple Reactors 模式通常也可以等同于 Master-Workers 模式，比如 Nginx 和 Memcached 等就是采用這種多執行緒模型，雖然不同的專案實作細節略有區別，但總體來說模式是一致的，

圖4-8

Acceptor，請求接收者，在實踐時其職責類似服務器，并不真正負責連接請求的建立，而只將其請求委托 Main Reactor 執行緒池來實作，起到一個轉發的作用，
Main Reactor，主 Reactor 執行緒組，主要負責連接事件，并將IO讀寫請求轉發到 SubReactor 執行緒池，
Sub Reactor，Main Reactor 通常監聽客戶端連接后會將通道的讀寫轉發到 Sub Reactor 執行緒池中一個執行緒(負載均衡)，負責資料的讀寫，在 NIO 中通常注冊通道的讀(OP_READ)、寫事件(OP_WRITE)，

MultiplyReactor

public class MultiplyReactor {
    public static void main(String[] args) throws IOException {
        MultiplyReactor mr = new MultiplyReactor(8888);
        mr.start();
    }
    private static final int POOL_SIZE = Runtime.getRuntime().availableProcessors();
    // Reactor（Selector） 執行緒池，其中一個執行緒被 mainReactor 使用，剩余執行緒都被 subReactor 使用
    static Executor mainReactorExecutor = Executors.newFixedThreadPool(POOL_SIZE);
    // 主 Reactor，接收連接，把 SocketChannel 注冊到從 Reactor 上
    private Reactor mainReactor;
    private int port;

    public MultiplyReactor(int port) {
        try {
            this.port = port;
            mainReactor = new Reactor();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * 啟動主從 Reactor，初始化并注冊 Acceptor 到主 Reactor
     */
    public void start() throws IOException {
        new Acceptor(mainReactor.getSelector(), port); // 將 ServerSocketChannel 注冊到 mainReactor
        mainReactorExecutor.execute(mainReactor); //使用執行緒池來處理main Reactor的連接請求
    }
}

Reactor

public class Reactor implements Runnable{
    private ConcurrentLinkedQueue<AsyncHandler> events=new ConcurrentLinkedQueue<>();
    private final Selector selector;

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

    public Selector getSelector(){
        return selector;
    }
    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                AsyncHandler handler;
                while ((handler = events.poll()) != null) {
                    handler.getChannel().configureBlocking(false);
                    SelectionKey sk=handler.getChannel().register(selector, SelectionKey.OP_READ);
                    sk.attach(handler);
                    handler.setSk(sk);
                }
                selector.select(); //阻塞
                Set<SelectionKey> selectionKeys=selector.selectedKeys();
                Iterator<SelectionKey> it=selectionKeys.iterator();
                while(it.hasNext()){
                    SelectionKey key=it.next();
                    //獲取attach方法傳入的附加物件
                    Runnable runnable=(Runnable)key.attachment();
                    if(runnable!=null){
                        runnable.run();
                    }
                    it.remove();
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    public void register(AsyncHandler asyncHandler){
        events.offer(asyncHandler);
        selector.wakeup();
    }
}

Acceptor

public class Acceptor implements Runnable{

    final Selector sel;
    final ServerSocketChannel serverSocket;
    int handleNext = 0;

    private final int POOL_SIZE=Runtime.getRuntime().availableProcessors();
    private Executor subReactorExecutor= Executors.newFixedThreadPool(POOL_SIZE);

    private Reactor[] subReactors=new Reactor[POOL_SIZE-1];

    public Acceptor(Selector sel, int port) throws IOException {
        this.sel = sel;
        serverSocket = ServerSocketChannel.open();
        serverSocket.socket().bind(new InetSocketAddress(port)); // 系結埠
        // 設定成非阻塞模式
        serverSocket.configureBlocking(false);
        // 注冊到 選擇器 并設定處理 socket 連接事件
        serverSocket.register(sel, SelectionKey.OP_ACCEPT,this);
        init();
        System.out.println("mainReactor-" + "Acceptor: Listening on port: " + port);
    }
    public void init() throws IOException {
        for (int i = 0; i < subReactors.length; i++) {
            subReactors[i]=new Reactor();
            subReactorExecutor.execute(subReactors[i]);
        }
    }
    @Override
    public synchronized void run() {
        try {
            // 接收連接，非阻塞模式下，沒有連接直接回傳 null
            SocketChannel sc = serverSocket.accept();
            if (sc != null) {
                // 把提示發到界面
                sc.write(ByteBuffer.wrap("Multiply Reactor Pattern Example\r\nreactor> ".getBytes()));
                System.out.println(Thread.currentThread().getName()+":Main-Reactor-Acceptor: " + sc.socket().getLocalSocketAddress() +" 注冊到 subReactor-" + handleNext);
                // 如何解決呢，直接呼叫 wakeup，有可能還沒有注冊成功又阻塞了，這是一個多執行緒同步的問題，可以借助佇列進行處理
                Reactor subReactor = subReactors[handleNext];
                subReactor.register(new AsyncHandler(sc));
                if(++handleNext == subReactors.length) {
                    handleNext = 0;
                }
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}

AsyncHandler

public class AsyncHandler implements Runnable{
    private SocketChannel channel;
    private SelectionKey sk;

    ByteBuffer inputBuffer=ByteBuffer.allocate(1024);
    ByteBuffer outputBuffer=ByteBuffer.allocate(1024);
    StringBuilder builder=new StringBuilder(); //存盤客戶端的完整訊息

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

    public SocketChannel getChannel() {
        return channel;
    }

    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){
            try {
                this.sk.channel().close();
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }

    protected void read() throws IOException {
        inputBuffer.clear();
        int n=channel.read(inputBuffer);
        if(inputBufferComplete(n)){
            System.out.println(Thread.currentThread().getName()+":Server端收到客戶端的請求訊息："+builder.toString());
            outputBuffer.put(builder.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{
                    builder.append((char)ch); //拼接讀取到的資料
                }
            }
        }else if(bytes==-1){
            throw new EOFException(); //客戶端關閉了連接
        }
        return false;
    }

    private void write() throws IOException {
        int written=-1;
        outputBuffer.flip(); //轉化為讀模式，判斷是否有資料需要發送
        if(outputBuffer.hasRemaining()){
            written=channel.write(outputBuffer); //把資料寫回客戶端
        }
        outputBuffer.clear();
        builder.delete(0,builder.length());
        if(written<=0){ //表示客戶端沒有輸資訊
            this.sk.channel().close();
        }else{
            channel.write(ByteBuffer.wrap("\r\nreactor>".getBytes()));
            this.sk.interestOps(SelectionKey.OP_READ);
        }
    }
}

關注[跟著Mic學架構]公眾號，獲取更多精品原創

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/322953.html

標籤：Java

上一篇：面試官：SOA 和微服務的區別？這回終于搞清楚了！

下一篇：一個簡單的超市訂單管理系統（4）-用戶注銷和權限攔截

從網路通信的演程序序徹底搞懂Redis高性能通信的原理（全網最詳細，建議收藏）

Redis 為什么那么快

從請求處理開始分析

網路IO的通信原理

NIO多路復用機制

BIO模型

構建基礎的BIO通信模型

基于多執行緒優化BIO

NIO非阻塞IO

多路復用IO

單執行緒Reactor 模型（高性能I/O設計模式）

Reactor

Acceptor

Handler

ReactorClient

多執行緒單Reactor模型

多執行緒改造-MultiDispatchHandler

Acceptor

多執行緒Reactor總結

多Reactor多執行緒模式（主從多Reactor模型）

MultiplyReactor

Reactor

Acceptor

AsyncHandler