reactor模型學習

學習鏈接

狂野架構師第四期netty視頻 - B站視頻
狂野架構師訓練營6期 - B站視頻

Netty學習example示例（含官方示例代碼）

LG-Netty學習

【硬核】肝了一月的Netty知識點 - 啟動過程寫的很詳細

Reactor模型講解
一文搞懂Reactor模型與實現
高性能網絡編程之 Reactor 網絡模型（徹底搞懂）

文章目錄

學習鏈接
BIO&NIO代碼實現
- BIO（傳統阻塞IO）
- - BioServer
  - BioClient
- NIO1（基礎理解，不交互）
- - NioServer
  - NioClient
- NIO2（交互，不處理write后續）
- - NioServer
  - NioClient
- NIO3（交互，處理write后續）
- - NioServer
  - NioClient
Reactor模型
- 單Reactor-單線程
- 單Reactor-多線程
- 主從Reactor-多線程
- - 工作流程
  - 優勢
Netty
- 介紹
- - 關于異步
  - Netty核心架構
  - 網絡通信框架為什么非得是Netty
  - Netty現狀如何
  - Netty對三種IO的支持
  - Netty中的Reactor實現
  - - 工作流程
    - 總結
- Pipeline 和 Handler
- - ChannelPipeline & ChannelHandler
  - ChannelHandler 分類
- Netty如何使用Reactor模式

BIO&NIO代碼實現

BIO（傳統阻塞IO）

BioServer

@Slf4j
public class BioServer {public static void main(String[] args) {//由Acceptor線程負責監聽客戶端的連接ServerSocket serverSocket = null;try {// 執行完，服務器啟動成功（在此阻塞，直到啟動成功或拋出異常）serverSocket = new ServerSocket(8888);System.out.println("服務端啟動監聽.......");while (true) {//Acceptor線程接收到客戶端連接請求之后為每個客戶端創建一個新的線程進行業務處理（在此阻塞，直到接收到1個客戶端連接或拋出異常）Socket socket = serverSocket.accept();System.out.println("成功接收一個客戶端連接:"+socket.getInetAddress());new Thread(new ServerHandler(socket)).start();}} catch (IOException e) {log.error("服務端發生異常", e);}finally {if (serverSocket!=null) {try {serverSocket.close();} catch (IOException e) {e.printStackTrace();}}}}
}@Slf4j
class ServerHandler implements  Runnable{private final Socket socket;public ServerHandler(Socket socket) {this.socket = socket;}public void run() {BufferedReader in = null;BufferedWriter out = null;try {//獲取客戶端的輸入流in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));out = new BufferedWriter(new OutputStreamWriter(this.socket.getOutputStream()));System.out.println("準備接收來自客戶端:"+this.socket.getInetAddress()+"的數據");//讀取客戶端發送過來的數據while (true) {// 當客戶端調用socket.close()時，line會為null；// 但當客戶端突然關閉（比如代碼執行完，客戶端直接退出），此時readLine()會拋出異常String line = in.readLine();if (line == null) {log.info("讀取內容是null");break;}System.out.println("成功接收來自客戶端的數據:"+ line);//進行業務處理//給客戶端響應數據out.write("success! i am server \n");out.flush();}} catch (IOException e) {if (in != null) {try {in.close();} catch (IOException ioException) {ioException.printStackTrace();}}if (out != null) {try {out.close();} catch (IOException ioException) {ioException.printStackTrace();}}}}
}

BioClient

public class BioClient {public static void main(String[] args) {Socket socket = null;BufferedReader in = null;BufferedWriter out = null;try {// 當 new Socket("127.0.0.1", 8080) 構造函數成功返回時，意味著客戶端Socket已經與服務器成功建立了TCP連接socket = new Socket("127.0.0.1",8888);in = new BufferedReader(new InputStreamReader(socket.getInputStream()));out = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));System.out.println("準備向服務端寫數據!");//向服務端寫數據out.write("hello server , i am client ! \n");//注意別丟 \n 因為服務端是readLineout.flush();//接收來自服務端的數據String line = in.readLine();System.out.println("成功接收到來自服務端的數據:"+line);// 可以選擇在此處給服務端1個友好的關閉信號// socket.close();} catch (IOException e) {if (in != null) {try {in.close();} catch (IOException ioException) {ioException.printStackTrace();}}if (out != null) {try {out.close();} catch (IOException ioException) {ioException.printStackTrace();}}if (socket != null) {try {socket.close();} catch (IOException ioException) {ioException.printStackTrace();}}}}
}

NIO1（基礎理解，不交互）

NioServer

public class NioServer {/*** 基于 Channel開發** @param args*/public static void main(String[] args) {try {//1、打開ServerSocketChannel,用于監聽客戶端的連接，它是所有客戶端連接的父管道(代表客戶端連接的管道都是通過它創建的)ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();//2、綁定監聽端口，設置連接為非阻塞模式（在非阻塞模式下，bind方法也是同步調用）// 只有（要）綁定后，才（就）能調用 serverSocketChannel.accept()serverSocketChannel.socket().bind(new InetSocketAddress(8888));// 此方法可以在任何時候調用。新的阻塞模式僅對在此方法返回之后發起的 I/O 操作生效。serverSocketChannel.configureBlocking(false);//3、創建多路復用器SelectorSelector selector = Selector.open();//4、將ServerSocketChannel注冊到selector上，監聽客戶端連接事件ACCEPTserverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);System.out.println("服務端已成功啟動,可以接收連接!");//5、創建 Reactor線程，讓多路復用器在 Reactor 線程中執行多路復用程序new Thread(new SingleReactor(selector)).start();} catch (IOException e) {e.printStackTrace();}}
}@Slf4j
class SingleReactor implements Runnable {private final Selector selector;public SingleReactor(Selector selector) {this.selector = selector;}public void run() {//6、selector輪詢準備就緒的事件while (true) {try {// 1、當客戶端突然退出，也會觸發可讀的就緒事件集，并且此時去讀，會拋出異常，//    如果此時忽略這個異常，那么下次select()查詢時，不會阻塞住，所以此時就需要取消這個keyselector.select(1000);// 獲取到所查詢到的感興趣的事件集Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()) {// 關于這個SelectionKey有幾個需要理解的點// 1. select()時，當注冊到selector的channel中發生了感興趣的事件時，就會返回代表channel和selector注冊關系的selectionKey，//    這個selectionKey就是channel注冊到selector時返回的，是同一對象。//    通過selectionKey可以直到是何種感興趣的事件，這個事件有可能是多個。// 2. 既然產生了感興趣的事件，那么這個事件就必須得到處理，否則下次select()查詢時，由于仍然有感興趣的事件，所以不會阻塞住，必須處理掉這個事件。// 3. 必須將當前處理的key從key集合中移除掉，假設處理了事件，但并不移除這個key，那么下次select()查詢會阻塞，但當其它channel上發生的感興趣事件時，//    此時就不阻塞了，然后再調用selector.selectedKeys()方法，會把上次沒有移除的這個key也給返回在key集合中，但是那個對應的事件實際上已經被處理了。//    所以這個selectedKeys集合，在jdk底層是不會幫我們自動移除的，它只會在注冊的channel發生了感興趣的事件時，會把這個channel對應的selectionKey放入到selectedKeys集合SelectionKey selectionKey = iterator.next();iterator.remove();try {processKey(selectionKey);} catch (IOException e) {log.error("處理selectionKey發生異常", e);// 1、當客戶端突然退出，也會觸發可讀的就緒事件集，并且此時去讀，會拋出異常，//    如果此時忽略這個異常（啥都不干，只是打印個日志），那么下次select()查詢時，不會阻塞住，所以此時就需要取消這個key，避免死循環selectionKey.cancel();SelectableChannel channel = selectionKey.channel();if (channel != null) {channel.close();}}}} catch (IOException e) {log.info("服務端發生異常", e);}}}private void processKey(SelectionKey key) throws IOException {if (key.isValid()) {//7、根據準備就緒的事件類型分別處理if (key.isAcceptable()) { //客戶端請求連接事件就緒//7.1、接收一個新的客戶端連接，創建對應的SocketChannel,ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();// 此處accept方法會以非阻塞方式執行SocketChannel socketChannel = serverSocketChannel.accept();//7.2、設置socketChannel的非阻塞模式，并將其注冊到Selector上，監聽讀事件// 只有非阻塞模式的socketChannel才能注冊到selector上socketChannel.configureBlocking(false);// 此處注冊時，可以指定攜帶1個附件socketChannel.register(this.selector, SelectionKey.OP_READ);}if (key.isReadable()) {//讀事件準備繼續//7.1、讀客戶端發送過來的數據SocketChannel socketChannel = (SocketChannel) key.channel();ByteBuffer readBuffer = ByteBuffer.allocate(1024);// 1、將socketChannel的數據讀到readBuffer中// 2、read仍然以非阻塞方式執行// 3、如果1次沒有讀完接收緩沖區中的數據，則下次仍然會觸發可讀事件，可以接著讀// 4、如果接收緩沖區中沒有數據，則立即返回0（比如說，我先一次全部讀完了，我又去調用這個方法去讀一遍，此時緩沖區中已經沒有數據可讀了）// 5、當客戶端給了1個關閉的信號時，會觸發readyOps就會事件為可讀，此時讀取會返回-1// 6、如果客戶端沒給關閉信號，就直接退出了，這時去讀就會拋出異常（比如客戶端執行完所有代碼就退出了，或者debug模式下直接強制關閉客戶端）int readBytes = socketChannel.read(readBuffer);//前面設置過socketChannel是非阻塞的，故要通過返回值判斷讀取到的字節數if (readBytes > 0) {readBuffer.flip();//讀寫模式切換byte[] bytes = new byte[readBuffer.remaining()];readBuffer.get(bytes);String msg = new String(bytes, "utf-8");//進行業務處理String response = doService(msg);//給客戶端響應數據System.out.println("服務端開始向客戶端響應數據");byte[] responseBytes = response.getBytes();ByteBuffer writeBuffer = ByteBuffer.allocate(responseBytes.length);writeBuffer.put(responseBytes);writeBuffer.flip();// 1、以非阻塞方式寫出數據給客戶端，意思就是發送緩沖區中當前能寫多少量，反正不可能超過這個量// 2、這里不一定能夠一次就能把writeBuffer中的數據全部寫給客戶端// 3、返回的int表示寫了多少// 4、所以如果沒寫完，還需監聽可寫事件，然后將未寫完的數據繼續寫出去socketChannel.write(writeBuffer);} else if (readBytes < 0) {//值為-1表示鏈路通道已經關閉key.cancel();socketChannel.close();} else {//沒讀取到數據，忽略log.warn("沒讀取到數據，忽略");}}}}private String doService(String msg) {System.out.println("成功接收來自客戶端發送過來的數據:" + msg);return msg + "---from nioserver";}}

NioClient

public class NioClient {public static void main(String[] args) {try {//1、窗口客戶端SocketChannel,綁定客戶端本地地址(不選默認隨機分配一個可用地址)SocketChannel socketChannel = SocketChannel.open();//2、設置非阻塞模式,socketChannel.configureBlocking(false);//3、創建SelectorSelector selector = Selector.open();//3、創建Reactor線程new Thread(new SingleReactorClient(socketChannel, selector)).start();} catch (IOException e) {e.printStackTrace();}}
}@Slf4j
class SingleReactorClient implements Runnable {private final SocketChannel socketChannel;private final Selector selector;public SingleReactorClient(SocketChannel socketChannel, Selector selector) {this.socketChannel = socketChannel;this.selector = selector;}public void run() {try {//連接服務端doConnect(socketChannel, selector);} catch (IOException e) {e.printStackTrace();System.exit(1);}//5、多路復用器執行多路復用程序while (true) {try {selector.select(1000);Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()) {SelectionKey selectionKey = iterator.next();processKey(selectionKey);iterator.remove();}} catch (IOException e) {log.info("nio client 異常", e);}}}private void doConnect(SocketChannel sc, Selector selector) throws IOException {System.out.println("客戶端成功啟動,開始連接服務端");//3、連接服務端boolean connect = sc.connect(new InetSocketAddress("127.0.0.1", 8888));//4、將socketChannel注冊到selector并判斷是否連接成功，連接成功監聽讀事件，沒有連接成功則繼續監聽連接事件System.out.println("connect=" + connect);if (connect) { // 如果連接的是本地，通常這里的connect是true了// 如果上來就連接成功了，則直接注冊讀事件（此時，就完全不需要連接事件了）sc.register(selector, SelectionKey.OP_READ);System.out.println("客戶端成功連上服務端,準備發送數據");//開始進行業務處理，向服務端發送數據（連接成功了，就（才）可以向socketChannel中寫入數據了）doService(sc);} else {// 如果上來就沒連接成功，則注冊連接事件sc.register(selector, SelectionKey.OP_CONNECT);}}private void processKey(SelectionKey key) throws IOException {if (key.isValid()) {//6、根據準備就緒的事件類型分別處理if (key.isConnectable()) {//服務端可連接事件準備就緒SocketChannel sc = (SocketChannel) key.channel();// 當selector監聽到連接結果后，就調用finishConnect確認連接是否成功// 當確認與客戶端連接成功后，才向服務端發送數據if (sc.finishConnect()) {//6.1、向selector注冊可讀事件(接收來自服務端的數據)// 注意這行代碼// 1、在監聽到連接成功后，此處操作同時暗含了取消了對連接事件的監聽，//    如果此處不取消對連接事件的監聽，則會一直觸發OP_CONNECT事件（經測試，后續一直觸發OP_CONNECT事件時，selector.selectedKeys()又無法獲取到對應的key）//    猜測原因：因為連接事件只會觸發一次，只處理一次，后續沒有再處理的必要了，所以不會再把僅關注OP_CONNECT事件的通道的key放到selectedKeys中了// 2、所以正確的處理步驟就是取消對連接事件的監聽，而監聽讀事件（寫事件需要在有東西可寫的時候再去關注）sc.register(selector, SelectionKey.OP_READ);//6.2、處理業務 向服務端發送數據doService(sc);} else {//連接失敗，退出System.exit(1);}}if (key.isReadable()) {//讀事件準備繼續//6.1、讀服務端返回的數據SocketChannel sc = (SocketChannel) key.channel();ByteBuffer readBufer = ByteBuffer.allocate(1024);// 要么是鏈路通道關閉了，這會觸發可讀事件，下一次select()將不會阻塞，因為鏈路通道關閉作為可讀事件一直觸發// 要么是讀到數據了但沒全讀完，下一次select()將不會阻塞，因為可讀事件一直觸發// 要么是數據全部一次就讀完了，// 要么讀取的時候拋出異常了int readBytes = sc.read(readBufer);//前面設置過socketChannel是非阻塞的，故要通過返回值判斷讀取到的字節數if (readBytes > 0) {readBufer.flip();//讀寫模式切換byte[] bytes = new byte[readBufer.remaining()];readBufer.get(bytes);String msg = new String(bytes, "utf-8");//接收到服務端返回的數據后進行相關操作doService(msg);} else if (readBytes < 0) {//值為-1表示鏈路通道已經關閉// 如果此時不取消該key，那么 鏈路通道關閉一直作為讀事件觸發，而引發不斷的循環key.cancel();sc.close();} else {//沒讀取到數據，忽略}}}}private static void doService(SocketChannel socketChannel) throws IOException {System.out.println("客戶端開始向服務端發送數據:");//向服務端發送數據byte[] bytes = "hello nioServer,i am nioClient !".getBytes();ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);writeBuffer.put(bytes);writeBuffer.flip();// 這里可能不能一次性寫完socketChannel.write(writeBuffer);}private String doService(String msg) {System.out.println("成功接收來自服務端響應的數據:" + msg);return "";}
}

NIO2（交互，不處理write后續）

添加交互，但不考慮是否能一次性寫完的問題。

NioServer

import lombok.extern.slf4j.Slf4j;import java.io.IOException;
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.util.Iterator;
import java.util.List;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;@Slf4j
public class NioServer {public static void main(String[] args) {try {ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();serverSocketChannel.configureBlocking(false);serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 8080));Selector selector = Selector.open();serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);new Thread(new SingleReactor(serverSocketChannel, selector)).start();} catch (IOException e) {log.info("服務器發生異常", e);}}}@Slf4j
class SingleReactor implements Runnable {private ServerSocketChannel serverSocketChannel;private Selector selector;private List<SocketChannel> channels = new CopyOnWriteArrayList<>();private AtomicBoolean terminated = new AtomicBoolean(false);public SingleReactor(ServerSocketChannel serverSocketChannel, Selector selector) {this.serverSocketChannel = serverSocketChannel;this.selector = selector;}@Overridepublic void run() {new Thread(() -> {Scanner sc = new Scanner(System.in);while (true) {log.info("請輸入：");String msg = sc.nextLine();if ("exit".equals(msg)) {terminated.set(true);try {// 【此處發現，當調用的serverSocketChannel的close()方法時，selector.select()不會停止阻塞】serverSocketChannel.close();} catch (IOException e) {log.error("關閉serverSocketChannel發生異常");}log.info("關閉服務 {}", channels.size());for (SocketChannel channel : channels) {try {// 【此處發現，當調用socketChannel的close()方法時，selector.select()會停止阻塞，相當于給了選擇器一個事件通知】channel.close();log.info("關閉客戶端");} catch (IOException e) {log.error("exit 關閉channel錯誤", e);}}break;} else {for (SocketChannel channel : channels) {try {channel.write(ByteBuffer.wrap(("服務器主動群發消息->" + msg).getBytes()));} catch (IOException e) {log.error("server發送數據發生異常", e);try {channel.close();} catch (IOException ex) {log.error("server關閉連接發生異常", ex);}}}}}}).start();while (!terminated.get()) {try {log.info("server 開始進入select");// 客戶端channel調用close 或者 服務端這邊的channel(SocketChannel，不包括ServerSocketChannel)調用close都會喚醒selectorselector.select();Set<SelectionKey> keys = selector.selectedKeys();log.info("server selectedKeys {}", keys.size());Iterator<SelectionKey> it = keys.iterator();while (it.hasNext()) {SelectionKey key = it.next();it.remove();log.info("selectionKey {}, {}", key, key.readyOps());try {if (key.isAcceptable()) {ServerSocketChannel ssc = (ServerSocketChannel) key.channel();SocketChannel socketChannel = ssc.accept();socketChannel.configureBlocking(false);socketChannel.register(selector, SelectionKey.OP_READ);log.info("接收到客戶端連接，并注冊可讀事件：{}", socketChannel);channels.add(socketChannel);}if (key.isReadable()) {SocketChannel socketChannel = (SocketChannel) key.channel();log.info("讀取客戶端 {} 的數據", socketChannel);ByteBuffer readBuffer = ByteBuffer.allocate(1024);int readBytes = socketChannel.read(readBuffer);log.info("讀取到 {} 字節數據", readBytes);if (readBytes > 0) {readBuffer.flip();byte[] bytes = new byte[readBuffer.remaining()];readBuffer.get(bytes);String msg = new String(bytes, "utf-8");log.info("接收到客戶端 {} 的數據：{}", socketChannel, msg);socketChannel.write(ByteBuffer.wrap(("服務器已收到消息->" + msg).getBytes()));} else if (readBytes == -1) {log.info("客戶端 {} 已關閉連接", socketChannel);key.cancel();socketChannel.close();channels.remove(socketChannel);} else {log.info("沒有數據可讀");}}} catch (IOException e) {log.error("處理key發生異常", e);key.cancel();key.channel().close();}}} catch (IOException e) {log.error("server發生異常", e);}}}
}

NioClient

import lombok.extern.slf4j.Slf4j;import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;@Slf4j
public class NioClient {public static void main(String[] args) {try {SocketChannel socketChannel = SocketChannel.open();socketChannel.configureBlocking(false);Selector selector = Selector.open();new Thread(new SingleReactorClient(selector, socketChannel)).start();} catch (IOException e) {log.info("客戶端拋出異常", e);}}}@Slf4j
class SingleReactorClient implements Runnable {private Selector selector;private SocketChannel socketChannel;private AtomicBoolean terminated = new AtomicBoolean(false);private Thread chatThread = null;public SingleReactorClient(Selector selector, SocketChannel socketChannel) {this.selector = selector;this.socketChannel = socketChannel;}@Overridepublic void run() {try  {boolean connected = socketChannel.connect(new InetSocketAddress("localhost", 8080));if (connected) {log.info("第一次就連接成功了");writeToServer("【連接成功111】", socketChannel);startChatThread();} else {log.info("注冊連接事件監聽");socketChannel.register(selector, SelectionKey.OP_CONNECT);}while (!terminated.get()) {log.info("開始進入select");// 打印當前所有注冊的鍵Set<SelectionKey> allKeys = selector.keys();log.info("當前注冊的鍵數量: {}", allKeys.size());for (SelectionKey k : allKeys) {log.info("鍵: {}, interestOps: {}, readyOps: {}", k, k.interestOps(), k.readyOps());}int selectedCount = selector.select();log.info("select返回，就緒通道數量: {}", selectedCount);Set<SelectionKey> keys = selector.selectedKeys();log.info("selectedKeys {}", keys.size());Iterator<SelectionKey> it = keys.iterator();while (it.hasNext()) {SelectionKey key = it.next();it.remove();log.info("selectionKey {}, {}", key, key.readyOps());if (key.isConnectable()) {if (socketChannel.finishConnect()) {log.info("監聽連接成功事件");socketChannel.register(selector, SelectionKey.OP_READ);writeToServer("【連接成功222】", socketChannel);startChatThread();} else {log.error("連接失敗222");return;}}if (key.isReadable()) {SocketChannel channel = (SocketChannel) key.channel();ByteBuffer readBufer = ByteBuffer.allocate(1024);int readBytes = channel.read(readBufer);if (readBytes > 0) {log.info("讀取到{}字節數據", readBytes);readBufer.flip();log.info("讀取數據：{}", new String(readBufer.array(), 0, readBytes));} else if (readBytes == -1) {log.info("客戶端 {} 已關閉連接", socketChannel);key.cancel();terminated.set(true);if (chatThread != null) {// 中斷并不能使得scanner的nextLine()方法停止阻塞，所以導致chatThread不能停止// 所以，也可以使用System.in.exit()chatThread.interrupt();}} else {log.info("沒有數據可讀");}}}}log.info("客戶端循環結束...");} catch (IOException e) {log.error("client發生異常", e);}log.info("客戶端停止...");}private void startChatThread() {chatThread = new Thread(new Runnable() {@Overridepublic void run() {while (!Thread.currentThread().isInterrupted()) {try {log.info("請輸入：");String msg = getInput();if ("exit".equals(msg)) {terminated.set(true);// 【此處發現，當調用socketChannel的close()方法時，selector.select()會停止阻塞，相當于給了選擇器一個事件通知】socketChannel.close();break;// 也可以選擇直接退出程序// System.exit(0);} else {writeToServer(msg, socketChannel);}} catch (IOException e) {log.error("client發送數據發生異常", e);try {socketChannel.close();} catch (IOException ex) {log.error("client關閉連接發生異常", ex);}}}}private String getInput() {InputStream in = System.in;while (!Thread.currentThread().isInterrupted()) {try {if (in.available() > 0) {Scanner sc = new Scanner(System.in);String line = sc.nextLine();return line;}Thread.sleep(100);} catch (IOException e) {throw new RuntimeException(e);} catch (InterruptedException e) {// 直接結束return "exit";}}return null;}});chatThread.start();}private static void writeToServer(String msg, SocketChannel socketChannel) throws IOException {byte[] bytes = msg.getBytes();ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);writeBuffer.put(bytes);writeBuffer.flip();socketChannel.write(writeBuffer);}
}

NIO3（交互，處理write后續）

NioServer

import lombok.Data;
import lombok.extern.slf4j.Slf4j;import java.io.IOException;
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.util.Iterator;
import java.util.List;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;@Slf4j
public class NioServer {public static void main(String[] args) {try {ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();serverSocketChannel.configureBlocking(false);serverSocketChannel.bind(new InetSocketAddress("127.0.0.1", 8080));Selector selector = Selector.open();serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);new Thread(new SingleReactor(serverSocketChannel, selector)).start();} catch (IOException e) {log.info("服務器發生異常", e);}}}@Slf4j
class SingleReactor implements Runnable {private ServerSocketChannel serverSocketChannel;private Selector selector;private List<SocketChannel> channels = new CopyOnWriteArrayList<>();private AtomicBoolean terminated = new AtomicBoolean(false);public SingleReactor(ServerSocketChannel serverSocketChannel, Selector selector) {this.serverSocketChannel = serverSocketChannel;this.selector = selector;}@Overridepublic void run() {new Thread(() -> {Scanner sc = new Scanner(System.in);while (true) {log.info("請輸入：");String msg = sc.nextLine();if ("exit".equals(msg)) {terminated.set(true);try {// 【此處發現，當調用的serverSocketChannel的close()方法時，selector.select()不會停止阻塞】serverSocketChannel.close();} catch (IOException e) {log.error("關閉serverSocketChannel發生異常");}log.info("關閉服務 {}", channels.size());for (SocketChannel channel : channels) {try {// 【此處發現，當調用socketChannel的close()方法時，selector.select()會停止阻塞，相當于給了選擇器一個事件通知】channel.close();log.info("關閉客戶端");} catch (IOException e) {log.error("exit 關閉channel錯誤", e);}}break;} else {for (SocketChannel channel : channels) {try {// 不直接去寫了，而是關注可寫事件// channel.write(ByteBuffer.wrap(("服務器主動群發消息->" + msg).getBytes()));ByteBuffer buffer = ByteBuffer.wrap(("服務器主動群發消息->" + msg).getBytes());SelectionKey key = channel.keyFor(selector);Session session = (Session) key.attachment();session.setData(buffer);// 如果selector正在select()，此時調用key.interestOps(ops)修改感興趣的事件集，// 將不會讓selector停止阻塞，對于感興趣的事件集的改變只有在下一次selector.select()的時候才會生效// 所以此時就需要代碼去主動喚醒selector.select()停止阻塞key.interestOps(SelectionKey.OP_WRITE);// 需要去喚醒selector.wakeup();} catch (Exception e) {log.error("server發送數據發生異常", e);try {channel.close();} catch (IOException ex) {log.error("server關閉連接發生異常", ex);}}}}}}).start();while (!terminated.get()) {try {log.info("server 開始進入select");// 客戶端channel調用close 或者 服務端這邊的channel(SocketChannel，不包括ServerSocketChannel)調用close都會喚醒selectorselector.select();Set<SelectionKey> keys = selector.selectedKeys();log.info("server selectedKeys {}", keys.size());Iterator<SelectionKey> it = keys.iterator();while (it.hasNext()) {SelectionKey key = it.next();it.remove();if (key.isValid()) {log.info("selectionKey {}, {}", key, key.readyOps());try {if (key.isAcceptable()) {ServerSocketChannel ssc = (ServerSocketChannel) key.channel();SocketChannel socketChannel = ssc.accept();socketChannel.configureBlocking(false);// 攜帶1個附件socketChannel.register(selector, SelectionKey.OP_READ, new Session());log.info("接收到客戶端連接，并注冊可讀事件：{}", socketChannel);channels.add(socketChannel);}if (key.isReadable()) {SocketChannel socketChannel = (SocketChannel) key.channel();log.info("讀取客戶端 {} 的數據", socketChannel);ByteBuffer readBuffer = ByteBuffer.allocate(1024);int readBytes = socketChannel.read(readBuffer);log.info("讀取到 {} 字節數據", readBytes);if (readBytes > 0) {readBuffer.flip();byte[] bytes = new byte[readBuffer.remaining()];readBuffer.get(bytes);String msg = new String(bytes, "utf-8");log.info("接收到客戶端 {} 的數據：{}", socketChannel, msg);// 不直接去寫了，而是關注可寫事件// socketChannel.write(ByteBuffer.wrap(("服務器已收到消息->" + msg).getBytes()));// 將數據放入session中，待到寫事件中處理ByteBuffer buffer = ByteBuffer.wrap(("服務器已收到消息->" + msg).getBytes());((Session) key.attachment()).setData(buffer);// 關注可寫事件的同時，取消了可讀事件的關注（也就意味著即使后續有可讀的數據，也得等把當前的數據寫完了，再去讀）key.interestOps(SelectionKey.OP_WRITE);} else if (readBytes == -1) {log.info("客戶端 {} 已關閉連接", socketChannel);key.cancel();socketChannel.close();channels.remove(socketChannel);} else {log.info("沒有數據可讀");}}// 因為isWritable會校驗key是否被取消，如果被取消，再調用isWritable，就會拋異常if (key.isValid() && key.isWritable()) {log.info("處理寫事件");Session session = (Session) key.attachment();ByteBuffer buffer = session.getData();if (buffer != null) {SocketChannel channel = (SocketChannel) key.channel();int written = channel.write(buffer);log.info("limit: {}，position: {}，寫了 :{}", buffer.limit(), buffer.position(), written);if (!buffer.hasRemaining()) {key.interestOps(SelectionKey.OP_READ);log.info("寫完了，繼續關注可讀事件");}}}} catch (IOException e) {log.error("處理key發生異常", e);key.cancel();key.channel().close();}}}} catch (IOException e) {log.error("server發生異常", e);}}}
}@Data
class Session {private ByteBuffer data;public Session() {}}

NioClient

import lombok.extern.slf4j.Slf4j;import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;@Slf4j
public class NioClient {public static void main(String[] args) {try {SocketChannel socketChannel = SocketChannel.open();socketChannel.configureBlocking(false);Selector selector = Selector.open();new Thread(new SingleReactorClient(selector, socketChannel)).start();} catch (IOException e) {log.info("客戶端拋出異常", e);}}}@Slf4j
class SingleReactorClient implements Runnable {private Selector selector;private SocketChannel socketChannel;private AtomicBoolean terminated = new AtomicBoolean(false);private Thread chatThread = null;public SingleReactorClient(Selector selector, SocketChannel socketChannel) {this.selector = selector;this.socketChannel = socketChannel;}@Overridepublic void run() {try  {boolean connected = socketChannel.connect(new InetSocketAddress("localhost", 8080));if (connected) {log.info("第一次就連接成功了");writeToServer("【連接成功111】", socketChannel);startChatThread();} else {log.info("注冊連接事件監聽");socketChannel.register(selector, SelectionKey.OP_CONNECT);}while (!terminated.get()) {log.info("開始進入select");// 打印當前所有注冊的鍵Set<SelectionKey> allKeys = selector.keys();log.info("當前注冊的鍵數量: {}", allKeys.size());for (SelectionKey k : allKeys) {log.info("鍵: {}, interestOps: {}, readyOps: {}", k, k.interestOps(), k.readyOps());}int selectedCount = selector.select();log.info("select返回，就緒通道數量: {}", selectedCount);Set<SelectionKey> keys = selector.selectedKeys();log.info("selectedKeys {}", keys.size());Iterator<SelectionKey> it = keys.iterator();while (it.hasNext()) {SelectionKey key = it.next();it.remove();log.info("selectionKey {}, {}", key, key.readyOps());if (key.isConnectable()) {if (socketChannel.finishConnect()) {log.info("監聽連接成功事件");socketChannel.register(selector, SelectionKey.OP_READ);writeToServer("【連接成功222】", socketChannel);startChatThread();} else {log.error("連接失敗222");return;}}if (key.isReadable()) {SocketChannel channel = (SocketChannel) key.channel();ByteBuffer readBufer = ByteBuffer.allocate(1024);int readBytes = channel.read(readBufer);if (readBytes > 0) {log.info("讀取到{}字節數據", readBytes);readBufer.flip();log.info("讀取數據：{}", new String(readBufer.array(), 0, readBytes));} else if (readBytes == -1) {log.info("客戶端 {} 已關閉連接", socketChannel);key.cancel();terminated.set(true);if (chatThread != null) {// 中斷并不能使得scanner的nextLine()方法停止阻塞，所以導致chatThread不能停止// 所以，也可以使用System.in.exit()chatThread.interrupt();}} else {log.info("沒有數據可讀");}}}}log.info("客戶端循環結束...");} catch (IOException e) {log.error("client發生異常", e);}log.info("客戶端停止...");}private void startChatThread() {chatThread = new Thread(new Runnable() {@Overridepublic void run() {while (!Thread.currentThread().isInterrupted()) {try {log.info("請輸入：");String msg = getInput();if ("exit".equals(msg)) {terminated.set(true);// 【此處發現，當調用socketChannel的close()方法時，selector.select()會停止阻塞，相當于給了選擇器一個事件通知】socketChannel.close();break;// 也可以選擇直接退出程序// System.exit(0);} else {writeToServer(msg, socketChannel);}} catch (IOException e) {log.error("client發送數據發生異常", e);try {socketChannel.close();} catch (IOException ex) {log.error("client關閉連接發生異常", ex);}}}}private String getInput() {InputStream in = System.in;while (!Thread.currentThread().isInterrupted()) {try {if (in.available() > 0) {Scanner sc = new Scanner(System.in);String line = sc.nextLine();return line;}Thread.sleep(100);} catch (IOException e) {throw new RuntimeException(e);} catch (InterruptedException e) {// 直接結束return "exit";}}return null;}});chatThread.start();}private static void writeToServer(String msg, SocketChannel socketChannel) throws IOException {byte[] bytes = msg.getBytes();ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);writeBuffer.put(bytes);writeBuffer.flip();socketChannel.write(writeBuffer);}
}

Reactor模型

Reactor線程模型不是Java專屬，也不是Netty專屬，它其實是一種并發編程模型，是一種思想，具有指導意義

Reactor模型中定義了三種角色

Reactor：負責監聽和分配事件，將I/O事件分派給對應的Handler。新的事件包含連接建立就緒、讀就緒、寫就緒等。
Acceptor：處理客戶端新連接，并分派請求到處理器鏈中。
Handler：將自身與事件綁定，執行非阻塞讀/寫任務，完成channel的讀入，完成處理業務邏輯后，負責將結果寫出channel

單Reactor-單線程

NIO下Reactor單線程，所有的接收連接，處理數據的相關操作都在一個線程中來完成，性能上有瓶頸
在這里插入圖片描述

在這里插入圖片描述

單Reactor-多線程

NIO下Reactor多線程，把比較耗時的數據的編解碼運算操作放入線程池中來執行，提升了性能但還不是最好的方式

在這里插入圖片描述

主從Reactor-多線程

主從Reactor多線程，主從多線程，對于服務器來說，接收客戶端的連接是比較重要的，因此將這部分操作單獨用線程去操作
在這里插入圖片描述

工作流程

這種模式的基本工作流程為：

1）Reactor 主線程 MainReactor 對象通過 select 監聽客戶端連接事件，收到事件后，通過 Acceptor 處理客戶端連接事件。

2）當 Acceptor 處理完客戶端連接事件之后（與客戶端建立好 Socket 連接），MainReactor 將連接分配給SubReactor。（即：MainReactor 只負責監聽客戶端連接請求，和客戶端建立連接之后將連接交由SubReactor 監聽后面的 IO 事件。）

3）SubReactor 將連接加入到自己的連接隊列進行監聽，并創建 Handler 對各種事件進行處理。

4）當連接上有新事件發生的時候，SubReactor 就會調用對應的 Handler 處理。

5）Handler 通過 read 從連接上讀取請求數據，將請求數據分發給 Worker 線程池進行業務處理。

6）Worker 線程池會分配獨立線程來完成真正的業務處理，并將處理結果返回給 Handler。Handler 通過send 向客戶端發送響應數據。

7）一個 MainReactor 可以對應多個 SubReactor，即一個 MainReactor 線程可以對應多個 SubReactor 線程

優勢

這種模式的優勢如下：

1）MainReactor 線程與 SubReactor 線程的數據交互簡單職責明確，MainReactor 線程只需要接收新連接，SubReactor 線程完成后續的業務處理。

2）MainReactor 線程與 SubReactor 線程的數據交互簡單， MainReactor 線程只需要把新連接傳SubReactor 線程，SubReactor 線程無需返回數據。

3）多個 SubReactor 線程能夠應對更高的并發請求。

這種模式的缺點是編程復雜度較高。但是由于其優點明顯，在許多項目中被廣泛使用，包括 Nginx、Memcached、Netty 等。

這種模式也被叫做服務器的 1+M+N 線程模式，即使用該模式開發的服務器包含一個（或多個，1 只是表示相對較少）連接建立線程+M 個 IO 線程+N 個業務處理線程。這是業界成熟的服務器程序設計模式。

Netty

介紹

Netty是由JBOSS提供的一個java開源框架，現為 Github上的獨立項目。Netty提供非阻塞的、事件驅動的網絡應用程序框架和工具，用以快速開發高性能、高可靠性的網絡服務器和客戶端程序。

本質：網絡應用程序框架
實現：異步、事件驅動
特性：高性能、可維護、快速開發
用途：開發服務器和客戶端

https://netty.io/

https://netty.io/wiki/user-guide-for-4.x.html#wiki-h2-3

關于異步

程同步、異步是相對的，在請求或執行過程中，如果會阻塞等待，就是同步操作，反之就是異步操作
在這里插入圖片描述

Netty核心架構

在這里插入圖片描述

核心：

可擴展的事件模型
統一的通信api，簡化了通信編碼
零拷貝機制與豐富的字節緩沖區

傳輸服務：

支持socket以及datagram（數據報）
http傳輸服務
In-VM Pipe （管道協議，是jvm的一種進程）

協議支持：

http 以及 websocket
SSL 安全套接字協議支持
Google Protobuf （序列化框架）
支持zlib、gzip壓縮
支持大文件的傳輸
RTSP（實時流傳輸協議，是TCP/IP協議體系中的一個應用層協議）
支持二進制協議并且提供了完整的單元測試

在這里插入圖片描述

網絡通信框架為什么非得是Netty

Apache Mina：和Netty是同一作者，但是推薦Netty，作者認為Netty是針對Mina的重新打造版本，解決了一些問題并提高了擴展性
Sun Grizzly：用得少、文檔少，更新少。
Apple Swift NIO、ACE 等：其他語言不作考慮
Cindy 等：生命周期不長
Tomcat、Jetty：還沒有獨立出來，另外他們有自己的網絡通信層實現，是為了專門針對servelet容器而做的，不具備通用性。
- 那tomcat在網絡通信層為什么不選擇Netty呢？主要是由于tomcat出現的比較早

Netty現狀如何

在這里插入圖片描述

使用Netty的典型項目

數據庫： Cassandra
大數據處理： Spark、Hadoop
Message Queue：RocketMQ
檢索： Elasticsearch
框架：gRPC、Apache Dubbo、Spring5(響應式編程WebFlux)
分布式協調器：ZooKeeper
工具類： async-http-clien
…

Netty對三種IO的支持

在這里插入圖片描述

Netty中的Reactor實現

Netty線程模型是基于Reactor模型實現的，對Reactor三種模式都有非常好的支持，并做了一定的改進，也非常的靈活，一般情況，在服務端會采用主從架構模型。
在這里插入圖片描述

工作流程

在這里插入圖片描述
1）Netty 抽象出兩組線程池：BossGroup 和 WorkerGroup，每個線程池中都有EventLoop 線程（可以是OIO,NIO,AIO）。BossGroup中的線程專門負責和客戶端建立連接，WorkerGroup 中的線程專門負責處理連接上的讀寫, EventLoopGroup 相當于一個事件循環組，這個組中含有多個事件循環

2）EventLoop 表示一個不斷循環的執行事件處理的線程，每個EventLoop 都包含一個 Selector，用于監聽注冊在其上的 Socket 網絡連接（Channel）。

3）每個 Boss EventLoop 中循環執行以下三個步驟：

select：輪訓注冊在其上的 ServerSocketChannel 的 accept 事件（OP_ACCEPT 事件）
processSelectedKeys：處理 accept 事件，與客戶端建立連接，生成一個SocketChannel，并將其注冊到某個 WorkerEventLoop 上的 Selector 上
runAllTasks：再去以此循環處理任務隊列中的其他任務

4）每個 Worker EventLoop 中循環執行以下三個步驟：