一、原生NIO存在的問題

NIO的類庫與API繁雜，需要熟練掌握Selector、ServerSocketChannel、SocketChannel、Bytebuffer等要求熟悉Java多線程編程和網絡編程開發工作量和難度大，例如客戶端面臨斷連重連、網絡閃斷、半包讀寫、失敗緩存、網絡擁塞和異常流的處理等。JDK NIO的BUG：例如Epoll Bug，它會導致Selector空輪詢，最終導致CPU占用100%，到JDK1.7還未被有效解決

二、Netty說明

Netty是由JBOSS提供的一個Java開源框架。Netty提供異步的、基于事件驅動的網絡應用程序框架，用以快速開發高性能、高可靠的網絡IO程序。Netty可以幫助你快速、簡單的開發一個網絡應用，相當于簡化和流程化了NIO的開發過程。Netty是目前最流行的NIO框架，在互聯網領域、大數據分布式計算領域、游戲行業、通信行業等獲得了廣泛的應用，Elasticsearch、Dubbo框架內部都采用了Netty

三、Netty線程模型介紹

線程模型分類：

• 傳統阻塞I/O服務模型；
• Reactor模式(反應器模型)

3種典型的Reactor(根據Reactor數量和處理資源線程池數量不同)：

• 1）單Reactor單線程

• 2）單Reactor多線程

• 3）主從Reactor多線程

Netty線程模式——主要基于主從Reactor（有多個Reactor）多線程模型做了一定的改進

• 傳統阻塞IO服務模型
  

• 1、模型特點

采用阻塞IO獲取輸入的數據 每個連接都需要獨立的線程完成數據的輸入、業務處理、數據返回

• 2、問題分析

當并發數很大時，會創建大量的線程，占用很大的系統資源，連接創建后，如果當前線程暫時沒有數據可讀，該線程會阻塞在read操作上，造成線程資源浪費

• 3、解決方案

基于IO復用模型：多個連接共用一個阻塞對象，應用程序只需要在一個阻塞對象等待，無需阻塞所有連接，當某個連接有新的數據可以處理時，操作系統通知應用程序，線程從阻塞狀態返回，開始進行業務處理?；诰€程池復用線程資源：不必為每一個連接創建線程，將連接完成后的業務處理任務分配給線程進行處理，一個線程可以處理多個連接的業務

• Reactor模式（反應器、分發者、通知者模式）

• 1、工作原理及說明
  

Reactor模式，通過一個或多個輸入同時傳遞給服務器處理的模式（基于事件驅動），服務器端程序處理傳入的多個請求，并將它們同步分派到相應的處理線程。 Reactor模式使用了IO復用監聽事件，受到事件后分發給某個線程（進程），網絡服務高并發處理的關鍵

• 2、核心組成

• Reactor：在一個單獨的線程中運行，負責監聽和分發事件，分發給適當的處理程序對IO事件作出反應

• Handlers：處理程序執行IO事件要完成的實際事件。Reactor通過調用適當的處理程序來響應IO事件，處理程序非阻塞操作。

• 單Reactor單線程

1、工作原理及說明

• Select是前面IO復用模型介紹的標準網絡編程API，可以實現應用程序通過一個阻塞對象監聽多路連接請求；
• Reactor對象通過Select監控客戶端請求事件，收到事件后通過Dispatch進行分發；
• 如果建立連接請求事件，則由Acceptor通過Accept處理連接請求，然后創建一個Handler對象處理連接完成后的后續業務處理；
• 如果不是建立連接事件，則Reactor會分發給調用連接對應的Handler來響應；
• Handler會完成Read—>業務處理—>Send的完整業務流程；

2、優缺點分析

• 優點：模型簡單，無多線程、進程通信、競爭的問題，全部由一個線程完成；
• 缺點：性能問題，只有一個線程無法發揮出多核CPU的性能，Handler在處理某連接業務時，整個進程無法處理其他連接事件，容易導致性能瓶頸；可靠性問題，線程意外中止，或者進入死循環，會導致整個系統通信模塊不可用，不能接收和處理外部信息，節點故障；
• 使用場景：客戶端數量有限，業務處理快捷（例如 Redis在業務處理的時間復雜度為O(1)的情況）

• 單Reactor多線程

1、工作原理及說明

• Reactor通過Select監控客戶端請求事件，收到事件后，通過dispatch進行分發；
• 如果是建立連接的請求，則由Acceptor通過accept處理連接請求，同時創建一個handler處理完成連接后的后續請求；
• 如果不是連接請求，則由Reactor分發調用連接對應的handler來處理；
  Handler只負責響應事件，不做具體的業務處理，通過read讀取數據后，會分發給后面的worker線程池中的某個線程處理業務，Worker線程池會分配獨立的線程處理真正的業務，并將結果返回給Handler。Handler收到響應后，通過send方法將結果反饋給client

2、優缺點分析

• 優點：可以充分的利用多核CPU的處理能力；
• 缺點：多線程數據共享、訪問操作比較復雜，reactor處理所有的事件的監聽和響應，在單線程運行，在高并發場景容易出現性能瓶頸

• 主從Reactor多線程

1、工作原理及說明

• Reactor主線程MainReactor對象通過select監聽連接事件，收到事件后，通過Acceptor處理連接事件；
• 當Acceptor處理連接事件后，MainReactor將創建好的連接分配給SubReactor；
• SubReactor將連接加入到連接隊列進行監聽，并創建Handler進行各種事件處理；
• 當有新事件發生時，SubReactor調用對應的Handler進行處理
• Handler通過read讀取數據，分發給后面的Worker線程池處理
• Worker線程池會分配獨立的Worker線程進行業務處理，并將結果返回
• Handler收到響應結果后，通過send方法將結果返回給client

2、優缺點分析

• 優點：父線程和子線程的職責明確，父線程只需要接收新連接，子線程完成后續業務處理；
• 優點：父線程與子線程的數據交互簡單，Reactor主線程是需要把新連接傳給子線程，子線程無需返回數據
• 缺點：編程復雜度較高

• Reactor模式小結

• 響應快，雖然Reactor本身是同步的，但不必為單個同步事件所阻塞 最大程度的避免了復雜的多線程及同步問題，避免了多線程/進程的切換開銷
• 擴展性好，可以方便的通過增加Reactor勢力個數充分利用CPU資源
• 復用性好，Reactor模型本身與具體事件處理邏輯無關，具有很高的復用性

四、Netty模型

• 工作原理及說明
  

Netty抽象出兩組線程池：

• BossGroup專門負責接收客戶端的連接；
• WorkerGroup專門負責網絡的讀寫

BossGroup和WorkerGroup的類型都是NioEventLoopGroup

NioEventLoopGroup相當于一個事件循環組，組中含有多個事件循環，每一個事件循環是NioEventLoop。NioEventLoop表示一個不斷循環的執行任務的過程，每個NioEventLoop都有一個selector，用于監聽綁定在其上的socket的網絡通信。NioEventLoopGroup可以含有多個線程，即可以含有多個NioEventLoop

每個boss NioEventLoop執行的步驟有3步

• 【1】輪詢accept事件

• 【2】處理accept事件，與client建立連接，生成NioSocketChannel，并將其注冊到某個worker的NioEventLoop上的selector

• 【3】處理任務隊列的任務，即runAllTasks

每個worker的NioEventLoop循環執行的步驟：

• 【1】輪詢read,write事件

• 【2】處理IO事件，在對應的NioSocketChannel處理

• 【3】處理任務隊列的任務，即runAllTasks

每個worker NioEventLoop處理業務時，會使用PipeLine(管道)，pipeline中包含了channel，即通過pipeline可以獲取對應通道，通道中維護了很多處理器

• 案例分析

要求：

Netty服務器在6668端口監聽，客戶端能發送消息給服務器“hello,服務器”|| 服務器可以回復消息給客戶端“hello，客戶端”

設計步驟：

在項目中導入Netty開發需要的包：netty-all-xxx.Final.jar
編寫服務器端及服務器業務處理器

import com.SF.NIO.NIOServer; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.channel.socket.nio.NioSocketChannel;//創建BossGroup 和 WorkerGroup /** * 說明 * 1、創建兩個線程組 bossGroup 和 workerGroup * * 2、bossGroup 只是處理連接請求，真正與客戶端的業務處理交給 workerGroup 完成 * * 3、兩個都是無限循環 * * 4、bossGroup 和 workerGroup 含有的子線程（NioEventLoop）的個數 * 默認以 CPU 內核數*2 * */ EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { //創建服務器端的啟動對象，可以配置啟動參數 ServerBootstrap bootstrap = new ServerBootstrap(); //使用鏈式編程進行設置 bootstrap.group(bossGroup, workerGroup) //設置兩個線程組 .channel(NioServerSocketChannel.class) //使用 NioServerSocketChannel 作為服務器的通道實現 .option(ChannelOption.SO_BACKLOG, 128) //設置線程隊列得到連接個數 .childOption(ChannelOption.SO_KEEPALIVE, true) //設置保持活動連接狀態 .childHandler(new ChannelInitializer() { //創建一個通道初始化對象 //給 pipeline 設置處理器 @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new NettyServerHandler()); } }); //給我們的 workerGroup 的EventLoop 對應的管道設置處理器 System.out.println("服務器 is ready !!!"); //綁定一個端口并同步，生成一個ChannelFuture對象 //啟動服務器并綁定端口 ChannelFuture cf = bootstrap.bind(6668).sync(); //對關閉通道進行監聽 cf.channel().closeFuture().sync(); }finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } public class NettyServer { public static void main(String[] args) throws InterruptedException { } import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelPipeline; import io.netty.util.CharsetUtil;import java.nio.charset.Charset;/*** 說明* 1、我們自定義的 handler 需要繼承 netty 規定的某一個 HandlerAdapter* 2、這時我們自定義的一個 handler 才能稱為一個 handler* */* //讀取數據的事件（這里我們可以讀取客戶端發送的消息） /** * 1、ChannelHandlerContext ctx:上下文對象，含有管道pipeline，通道channel，地址 * 2、Object msg:即客戶端發送的數據，默認是Object * */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // super.channelRead(ctx, msg); System.out.println("服務器讀取線程 "+Thread.currentThread().getName()); System.out.println("server ctx = "+ctx); System.out.println("**********************************************************"); Channel channel = ctx.channel(); ChannelPipeline pipeline = ctx.pipeline(); //本質是一個雙向連表 //將msg轉成一個byteBuf //ByteBuf 是由 Netty 提供的，不是 NIO 的 ByteBuffer ByteBuf buf = (ByteBuf) msg; System.out.println("客戶端發送消息是："+buf.toString(CharsetUtil.UTF_8)); System.out.println("客戶端地址："+ctx.channel().remoteAddress()); } //數據讀取完畢 @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { // super.channelReadComplete(ctx); //將數據寫入到緩沖并刷新 //一般將發送的數據進行編碼 ctx.writeAndFlush(Unpooled.copiedBuffer("hello，客戶端~",CharsetUtil.UTF_8)); } //處理異常，一般需要關閉通道 @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { // super.exceptionCaught(ctx, cause); ctx.close(); } public class NettyServerHandler extends ChannelInboundHandlerAdapter { }

五、Netty模型小結

Netty抽象出兩組線程池：

• BossGroup專門負責接收客戶端的連接；
• WorkerGroup專門負責網絡的讀寫；

NioEventLoop表示一個不斷循環的執行任務的線程，每個NioEventLoop都有一個selector，用于監聽綁定在其上的socket的網絡通道；NioEventLoop內部采用串行化設計，從消息讀取->處理->編碼->發送始終由IO線程NioEventLoop負責。NioEventLoopGroup下包含多個NioEventLoop，每個NioEventLoop中包含一個Selector，一個taskQueue

• 每個NioEventLoop的Selector可以注冊監聽多個NioChannel
• 每個NioChannel只會綁定唯一的NioEventLoop
• 每個NioChannel都會綁定一個自己的ChannelPipeLine

文章轉自

總結

以上是生活随笔為你收集整理的Netty出现的原因以及多种Reactor模式的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。