文章目錄

• 一、 Netty 模型代碼解析
• 二、 Netty 案例服務(wù)器端代碼
• • 1 . 服務(wù)器主程序
  • 2 . 服務(wù)器自定義 Handler 處理者
• 三、 Netty 案例客戶端代碼
• • 1 . 客戶端主程序
  • 2 . 客戶端自定義 Handler 處理者
• 四、 Netty 案例運(yùn)行

一、 Netty 模型代碼解析

---

1 . 線程池 NioEventLoopGroup :

① NioEventLoopGroup 線程池使用場(chǎng)景 : Netty 模型中的 BossGroup 和 WorkerGroup 都是 NioEventLoopGroup 類型的線程池 ;

② NioEventLoopGroup 默認(rèn)線程個(gè)數(shù) : 系統(tǒng)默認(rèn)每個(gè)線程池中的 NioEventLoop 線程數(shù)是 CPU 核數(shù) $\times$ 2 , 下面的代碼可以獲取運(yùn)行 Netty 程序的設(shè)備的 CPU 核數(shù) ;

// 獲取設(shè)備的 CPU 核數(shù) NettyRuntime.availableProcessors()

③ 指定 NioEventLoopGroup 線程個(gè)數(shù) : 如果不想使用 Netty 線程池的默認(rèn)線程個(gè)數(shù) , 可以在 NioEventLoopGroup 構(gòu)造函數(shù)中子星設(shè)定線程數(shù) ;

// BossGroup 線程池 : 負(fù)責(zé)客戶端的連接 // 指定線程個(gè)數(shù) : 客戶端個(gè)數(shù)很少, 不用很多線程維護(hù), 這里指定線程池中線程個(gè)數(shù)為 1 EventLoopGroup bossGroup = new NioEventLoopGroup(1);

2 . NioEventLoopGroup 線程池線程分配 :

以客戶端連接完成后 , 數(shù)據(jù)讀寫場(chǎng)景舉例 ;

在 雙核 CPU 的服務(wù)器上 , NioEventLoopGroup 默認(rèn)有 $4$ 個(gè)線程 ; 按照順序循環(huán)分配 , 為第 $n$ 個(gè)客戶端分配第 $n\%4$ 個(gè) NioEventLoop 線程 ;

• 客戶端 $0$ 與服務(wù)器進(jìn)行數(shù)據(jù)交互在 NioEventLoop $0$ 線程中 ;
• 客戶端 $1$ 與服務(wù)器進(jìn)行數(shù)據(jù)交互在 NioEventLoop $1$ 線程中 ;
• 客戶端 $2$ 與服務(wù)器進(jìn)行數(shù)據(jù)交互在 NioEventLoop $2$ 線程中 ;
• 客戶端 $3$ 與服務(wù)器進(jìn)行數(shù)據(jù)交互在 NioEventLoop $3$ 線程中 ;
• 客戶端 $4$ 與服務(wù)器進(jìn)行數(shù)據(jù)交互在 NioEventLoop $0$ 線程中 ;
• 客戶端 $5$ 與服務(wù)器進(jìn)行數(shù)據(jù)交互在 NioEventLoop $1$ 線程中 ;
• 客戶端 $6$ 與服務(wù)器進(jìn)行數(shù)據(jù)交互在 NioEventLoop $2$ 線程中 ;
• 客戶端 $7$ 與服務(wù)器進(jìn)行數(shù)據(jù)交互在 NioEventLoop $3$ 線程中 ;

3 . NioEventLoopGroup 線程池封裝內(nèi)容 :

① NioEventLoopGroup 中的若干個(gè) NioEventLoop 線程都封裝在 children 中 , 線程個(gè)數(shù)是 CPU 核數(shù) 2 倍 ;

② 每個(gè) NioEventLoop 線程中封裝了如下內(nèi)容 :

• 選擇器 ( Selector ) , 用于監(jiān)聽客戶端的讀寫 IO 事件 ;
• 任務(wù)隊(duì)列 ( taskQueue ) , 用于存儲(chǔ)事件對(duì)應(yīng)的業(yè)務(wù)邏輯任務(wù) ;
• 線程執(zhí)行器 ( executor ) , 用于執(zhí)行線程 ;

4 . ChannelHandlerContext 通道處理者上下文對(duì)象封裝內(nèi)容 :

① 用戶自定義的 處理者 ( Handler ) , 這里指的是 服務(wù)器端的 ServerHandr ( 自定義 ) , 客戶端的 ClientHandler ( 自定義 ) ;

② 管道 ( ChannelPipeline ) : 其本質(zhì)是雙向鏈表 , 該 ChannelHandlerContext 可以獲取該鏈表的前一個(gè) ( prev ) , 后一個(gè)管道對(duì)象 ( next ) ;

③ 管道 與 通道 :

• 二者都可以通過 通道處理者上下文 ( ChannelHandlerContext ) 獲取 ;
• 管道 與 通道 都可以互相從對(duì)方獲取 ;

Channel channel = ctx.channel(); ChannelPipeline pipeline = ctx.pipeline(); channel = pipeline.channel(); pipeline = channel.pipeline();

④ 管道 ( Pipeline ) 與 通道 ( Channel ) 關(guān)聯(lián) : 通過管道可以獲取通道 , 通過通道也可以獲取其對(duì)應(yīng)的管道 ;

5 . 處理者 ( Handler ) :

① 設(shè)置 Handler : 給 WorkerGroup 線程池中的 EventLoop 線程對(duì)應(yīng)的管道設(shè)置 處理器 ( Handler ) ;

② 自定義 Handler : 一般這個(gè) Handler 都是用戶自定義的類 , 繼承 ChannelInboundHandlerAdapter 類 ;

③ 運(yùn)行機(jī)制 : 在 BossGroup 中連接客戶端成功后 , 將 NioSocketChannel 注冊(cè)給 WorkerGroup 中的 EventLoop 中的 選擇器 ( Selector ) , 如果監(jiān)聽到客戶端數(shù)據(jù) IO 事件 , 就會(huì)調(diào)用 管道 ( Pipeline ) 處理該事件 , 管道 ( Pipeline ) 中調(diào)用 處理器 ( Handler ) 處理相應(yīng)的事件 , 該 處理器 ( Handler ) 可以是 Netty 提供的 , 也可以是開發(fā)者自定義的 ;

特別注意 : 自定義 Handler 中 , 重寫的 ChannelInboundHandlerAdapter 方法 , 將 super() 語句都刪除 ;

二、 Netty 案例服務(wù)器端代碼

---

1 . 服務(wù)器主程序

package kim.hsl.netty;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;/*** Netty 案例服務(wù)器端*/ public class Server {public static void main(String[] args) {// 1. 創(chuàng)建 BossGroup 線程池 和 WorkerGroup 線程池, 其中維護(hù) NioEventLoop 線程// NioEventLoop 線程中執(zhí)行無限循環(huán)操作// BossGroup 線程池 : 負(fù)責(zé)客戶端的連接// 指定線程個(gè)數(shù) : 客戶端個(gè)數(shù)很少, 不用很多線程維護(hù), 這里指定線程池中線程個(gè)數(shù)為 1EventLoopGroup bossGroup = new NioEventLoopGroup(1);// WorkerGroup 線程池 : 負(fù)責(zé)客戶端連接的數(shù)據(jù)讀寫EventLoopGroup workerGroup = new NioEventLoopGroup();// 2. 服務(wù)器啟動(dòng)對(duì)象, 需要為該對(duì)象配置各種參數(shù)ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup) // 設(shè)置 主從 線程組 , 分別對(duì)應(yīng) 主 Reactor 和 從 Reactor.channel(NioServerSocketChannel.class) // 設(shè)置 NIO 網(wǎng)絡(luò)套接字通道類型.option(ChannelOption.SO_BACKLOG, 128) // 設(shè)置線程隊(duì)列維護(hù)的連接個(gè)數(shù).childOption(ChannelOption.SO_KEEPALIVE, true) // 設(shè)置連接狀態(tài)行為, 保持連接狀態(tài).childHandler( // 為 WorkerGroup 線程池對(duì)應(yīng)的 NioEventLoop 設(shè)置對(duì)應(yīng)的事件 處理器 Handlernew ChannelInitializer<SocketChannel>() {// 創(chuàng)建通道初始化對(duì)象@Overrideprotected void initChannel(SocketChannel ch) throws Exception {// 為 管道 Pipeline 設(shè)置處理器 Hanedlerch.pipeline().addLast(new ServerHandr());}});System.out.println("服務(wù)器準(zhǔn)備完畢 ...");ChannelFuture cf = null;try {// 綁定本地端口, 進(jìn)行同步操作 , 并返回 ChannelFuturecf = bootstrap.bind(8888).sync();System.out.println("服務(wù)器開始監(jiān)聽 8888 端口 ...");// 關(guān)閉通道 , 開始監(jiān)聽操作cf.channel().closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();} finally {// 出現(xiàn)異常后, 優(yōu)雅的關(guān)閉bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}} }

2 . 服務(wù)器自定義 Handler 處理者

package kim.hsl.netty;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;/*** Handler 處理者, 是 NioEventLoop 線程中處理業(yè)務(wù)邏輯的類** 繼承 : 該業(yè)務(wù)邏輯處理者 ( Handler ) 必須繼承 Netty 中的 ChannelInboundHandlerAdapter 類* 才可以設(shè)置給 NioEventLoop 線程** 規(guī)范 : 該 Handler 類中需要按照業(yè)務(wù)邏輯處理規(guī)范進(jìn)行開發(fā)*/ public class ServerHandr extends ChannelInboundHandlerAdapter {/*** 讀取數(shù)據(jù) : 在服務(wù)器端讀取客戶端發(fā)送的數(shù)據(jù)* @param ctx* 通道處理者上下文對(duì)象 : 封裝了 管道 ( Pipeline ) , 通道 ( Channel ), 客戶端地址信息* 管道 ( Pipeline ) : 注重業(yè)務(wù)邏輯處理 , 可以關(guān)聯(lián)很多 Handler* 通道 ( Channel ) : 注重?cái)?shù)據(jù)讀寫* @param msg* 客戶端上傳的數(shù)據(jù)* @throws Exception*/@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {// 查看 ChannelHandlerContext 中封裝的內(nèi)容System.out.println("channelRead : ChannelHandlerContext ctx = " + ctx);// 將客戶端上傳的數(shù)據(jù)轉(zhuǎn)為 ByteBuffer// 這里注意該類是 Netty 中的 io.netty.buffer.ByteBuf 類// 不是 NIO 中的 ByteBuffer// io.netty.buffer.ByteBuf 性能高于 java.nio.ByteBufferByteBuf byteBuf = (ByteBuf) msg;// 將 ByteBuf 緩沖區(qū)數(shù)據(jù)轉(zhuǎn)為字符串, 打印出來System.out.println(ctx.channel().remoteAddress() + " 接收到客戶端發(fā)送的數(shù)據(jù) : " + byteBuf.toString(CharsetUtil.UTF_8));}/*** 服務(wù)器端讀取數(shù)據(jù)完畢后回調(diào)的方法* @param ctx* 通道處理者上下文對(duì)象 : 封裝了 管道 ( Pipeline ) , 通道 ( Channel ), 客戶端地址信息* * 管道 ( Pipeline ) : 注重業(yè)務(wù)邏輯處理 , 可以關(guān)聯(lián)很多 Handler* * 通道 ( Channel ) : 注重?cái)?shù)據(jù)讀寫* @throws Exception*/@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {// 數(shù)據(jù)編碼 : 將字符串編碼, 存儲(chǔ)到 io.netty.buffer.ByteBuf 緩沖區(qū)中ByteBuf byteBuf = Unpooled.copiedBuffer("Hello Client", CharsetUtil.UTF_8);// 寫出并刷新操作 : 寫出數(shù)據(jù)到通道的緩沖區(qū) ( write ), 并執(zhí)行刷新操作 ( flush )ctx.writeAndFlush(byteBuf);}/*** 異常處理 , 上面的方法中都拋出了 Exception 異常, 在該方法中進(jìn)行異常處理* @param ctx* @param cause* @throws Exception*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println("通道異常, 關(guān)閉通道");//如果出現(xiàn)異常, 就關(guān)閉該通道ctx.close();} }

三、 Netty 案例客戶端代碼

---

1 . 客戶端主程序

package kim.hsl.netty;import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel;public class Client {public static void main(String[] args) {// 客戶端只需要一個(gè) 時(shí)間循環(huán)組 , 即 NioEventLoopGroup 線程池EventLoopGroup eventLoopGroup = new NioEventLoopGroup();// 客戶端啟動(dòng)對(duì)象Bootstrap bootstrap = new Bootstrap();// 設(shè)置相關(guān)參數(shù)bootstrap.group(eventLoopGroup) // 設(shè)置客戶端的線程池.channel(NioSocketChannel.class) // 設(shè)置客戶端網(wǎng)絡(luò)套接字通道類型.handler( // 設(shè)置客戶端的線程池對(duì)應(yīng)的 NioEventLoop 設(shè)置對(duì)應(yīng)的事件處理器 Handlernew ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new ClientHandr());}});try {// 開始連接服務(wù)器, 并進(jìn)行同步操作// ChannelFuture 類分析 , Netty 異步模型// sync 作用是該方法不會(huì)再次阻塞ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8888).sync();System.out.println("客戶端連接服務(wù)器成功 ...");// 關(guān)閉通道, 開始監(jiān)聽channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();}finally {// 優(yōu)雅的關(guān)閉eventLoopGroup.shutdownGracefully();}} }

2 . 客戶端自定義 Handler 處理者

package kim.hsl.netty;import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.util.CharsetUtil;/*** Handler 處理者, 是 NioEventLoop 線程中處理業(yè)務(wù)邏輯的類** 繼承 : 該業(yè)務(wù)邏輯處理者 ( Handler ) 必須繼承 Netty 中的 ChannelInboundHandlerAdapter 類* 才可以設(shè)置給 NioEventLoop 線程** 規(guī)范 : 該 Handler 類中需要按照業(yè)務(wù)邏輯處理規(guī)范進(jìn)行開發(fā)*/ public class ClientHandr extends ChannelInboundHandlerAdapter {/*** 通道就緒后觸發(fā)該方法* @param ctx* @throws Exception*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 查看 ChannelHandlerContext 中封裝的內(nèi)容System.out.println("channelActive : ChannelHandlerContext ctx = " + ctx);// 數(shù)據(jù)編碼 : 將字符串編碼, 存儲(chǔ)到 io.netty.buffer.ByteBuf 緩沖區(qū)中ByteBuf byteBuf = Unpooled.copiedBuffer("Hello Server", CharsetUtil.UTF_8);// 寫出并刷新操作 : 寫出數(shù)據(jù)到通道的緩沖區(qū) ( write ), 并執(zhí)行刷新操作 ( flush )ctx.writeAndFlush(byteBuf);System.out.println("客戶端向服務(wù)器端發(fā)送 Hello Server 成功");}/*** 讀取數(shù)據(jù) : 在服務(wù)器端讀取客戶端發(fā)送的數(shù)據(jù)* @param ctx* 通道處理者上下文對(duì)象 : 封裝了 管道 ( Pipeline ) , 通道 ( Channel ), 客戶端地址信息* 管道 ( Pipeline ) : 注重業(yè)務(wù)邏輯處理 , 可以關(guān)聯(lián)很多 Handler* 通道 ( Channel ) : 注重?cái)?shù)據(jù)讀寫* @param msg* 服務(wù)器返回的數(shù)據(jù)* @throws Exception*/@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {// 查看 ChannelHandlerContext 中封裝的內(nèi)容System.out.println("channelRead : ChannelHandlerContext ctx = " + ctx);// 將服務(wù)器下發(fā)的數(shù)據(jù)轉(zhuǎn)為 ByteBuffer// 這里注意該類是 Netty 中的 io.netty.buffer.ByteBuf 類// 不是 NIO 中的 ByteBuffer// io.netty.buffer.ByteBuf 性能高于 java.nio.ByteBufferByteBuf byteBuf = (ByteBuf) msg;// 將 ByteBuf 緩沖區(qū)數(shù)據(jù)轉(zhuǎn)為字符串, 打印出來System.out.println(ctx.channel().remoteAddress() + " 服務(wù)器返回的數(shù)據(jù) : " + byteBuf.toString(CharsetUtil.UTF_8));}/*** 異常處理 , 上面的方法中都拋出了 Exception 異常, 在該方法中進(jìn)行異常處理* @param ctx* @param cause* @throws Exception*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println("通道異常, 關(guān)閉通道");//如果出現(xiàn)異常, 就關(guān)閉該通道ctx.close();} }

四、 Netty 案例運(yùn)行

---

1 . 運(yùn)行服務(wù)器端 : 服務(wù)器啟動(dòng) , 監(jiān)聽 8888 端口 ;

2 . 運(yùn)行客戶端 : 客戶端連接服務(wù)器的 8888 端口 , 并向服務(wù)器端寫出 Hello Server 字符串 , 之后便接到服務(wù)器端回送的 Hello Client 字符串信息 ;

3 . 查看客戶端 : 服務(wù)器端接收到客戶端信息 , 向客戶端寫出 Hello Client 字符串 ;

總結(jié)

以上是生活随笔為你收集整理的【Netty】Netty 入门案例分析 ( Netty 模型解析 | Netty 服务器端代码 | Netty 客户端代码 )的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。