[[358902]]

上文對(duì)IO模型和Reactor模型進(jìn)行講解，是不是感覺(jué)有點(diǎn)懵懵的。哈哈哈，反正我并沒(méi)有對(duì)其有深入見(jiàn)解。我是這樣安慰自己的，知識(shí)在不斷的反復(fù)學(xué)習(xí)和思考中有新的感悟。不氣餒，繼續(xù)新的征程。本篇文章想來(lái)從實(shí)戰(zhàn)開(kāi)始，帶我深入了解Netty各個(gè)組件是做什么?ByteBuf執(zhí)行原理又是怎樣的?

01一 第一個(gè)Netty實(shí)例

用Netty實(shí)現(xiàn)通信。說(shuō)白了就是客戶端向服務(wù)端發(fā)消息，服務(wù)端接收消息并給客戶端響應(yīng)。所以我來(lái)看看服務(wù)端和客戶端是如何實(shí)現(xiàn)的?

11.1 服務(wù)端

1. 依賴

<?xml version="1.0" encoding="UTF-8"?> 
<project xmlns="http://maven.apache.org/POM/4.0.0" 
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 
    <modelVersion>4.0.0</modelVersion> 
 
    <groupId>com.haopt.iot</groupId> 
    <artifactId>first-netty</artifactId> 
    <packaging>jar</packaging> 
    <version>1.0-SNAPSHOT</version> 
 
    <dependencies> 
        <dependency> 
            <groupId>io.netty</groupId> 
            <artifactId>netty-all</artifactId> 
            <version>4.1.50.Final</version> 
        </dependency> 
 
        <dependency> 
            <groupId>junit</groupId> 
            <artifactId>junit</artifactId> 
            <version>4.12</version> 
        </dependency> 
    </dependencies> 
 
    <build> 
        <plugins> 
             
            <plugin> 
                <groupId>org.apache.maven.plugins</groupId> 
                <artifactId>maven-compiler-plugin</artifactId> 
                <version>3.2</version> 
                <configuration> 
                    <source>1.8</source> 
                    <target>1.8</target> 
                    <encoding>UTF-8</encoding> 
                </configuration> 
            </plugin> 
        </plugins> 
    </build> 
</project> 

2. 服務(wù)端-MyRPCServer

package com.haopt.netty.server; 
 
import io.netty.bootstrap.ServerBootstrap; 
import io.netty.buffer.UnpooledByteBufAllocator; 
import io.netty.channel.ChannelFuture; 
import io.netty.channel.ChannelOption; 
import io.netty.channel.EventLoopGroup; 
import io.netty.channel.nio.NioEventLoopGroup; 
import io.netty.channel.socket.nio.NioServerSocketChannel; 
 
public class MyRPCServer { 
    public void start(int port) throws Exception { 
        // 主線程，不處理任何業(yè)務(wù)邏輯，只是接收客戶的連接請(qǐng)求 
        EventLoopGroup boss = new NioEventLoopGroup(1); 
        // 工作線程，線程數(shù)默認(rèn)是：cpu核數(shù)*2 
        EventLoopGroup worker = new NioEventLoopGroup(); 
        try { 
            // 服務(wù)器啟動(dòng)類 
            ServerBootstrap serverBootstrap = new ServerBootstrap(); 
            serverBootstrap.group(boss, worker) //設(shè)置線程組 
                    .channel(NioServerSocketChannel.class)  //配置server通道 
                    .childHandler(new MyChannelInitializer()); //worker線程的處理器 
            //ByteBuf 的分配要設(shè)置為非池化，否則不能切換到堆緩沖區(qū)模式 
            serverBootstrap.childOption(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT); 
            ChannelFuture future = serverBootstrap.bind(port).sync(); 
            System.out.println("服務(wù)器啟動(dòng)完成，端口為：" + port); 
            //等待服務(wù)端監(jiān)聽(tīng)端口關(guān)閉 
            future.channel().closeFuture().sync(); 
        } finally { 
            //優(yōu)雅關(guān)閉 
            boss.shutdownGracefully(); 
            worker.shutdownGracefully(); 
        } 
    } 
 
} 

3. 服務(wù)端-ChannelHandler

package com.haopt.netty.server.handler; 
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; 
 
public class MyChannelHandler extends ChannelInboundHandlerAdapter { 
    /** 
    * 獲取客戶端發(fā)來(lái)的數(shù)據(jù) 
    * @param ctx 
    * @param msg 
    * @throws Exception 
    */ 
    @Override 
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
        ByteBuf byteBuf = (ByteBuf) msg; 
        String msgStr = byteBuf.toString(CharsetUtil.UTF_8); 
        System.out.println("客戶端發(fā)來(lái)數(shù)據(jù)：" + msgStr); 
        //向客戶端發(fā)送數(shù)據(jù) 
        ctx.writeAndFlush(Unpooled.copiedBuffer("ok", CharsetUtil.UTF_8)); 
    } 
     
    /** 
    * 異常處理 
    * @param ctx 
    * @param cause 
    * @throws Exception 
    */ 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
        cause.printStackTrace(); 
        ctx.close(); 
    } 
} 

4. 測(cè)試用例

package com.haopt.netty.myrpc; 
import com.haopt.netty.server.MyRPCServer; 
import org.junit.Test; 
public class TestServer { 
    @Test 
    public void testServer() throws Exception{ 
        MyRPCServer myRPCServer = new MyRPCServer(); 
        myRPCServer.start(5566); 
    } 
} 

21.2 客戶端

1. 客戶端-client

package com.haopt.netty.client; 
import com.haopt.netty.client.handler.MyClientHandler; 
import io.netty.bootstrap.Bootstrap; 
import io.netty.channel.ChannelFuture; 
import io.netty.channel.EventLoopGroup; 
import io.netty.channel.nio.NioEventLoopGroup; 
import io.netty.channel.socket.nio.NioSocketChannel; 
public class MyRPCClient { 
    public void start(String host, int port) throws Exception { 
        //定義⼯作線程組 
        EventLoopGroup worker = new NioEventLoopGroup(); 
        try { 
            //注意：client使⽤的是Bootstrap 
            Bootstrap bootstrap = new Bootstrap(); 
            bootstrap.group(worker) 
            .channel(NioSocketChannel.class) //注意：client使⽤的是NioSocketChannel 
            .handler(new MyClientHandler()); 
            //連接到遠(yuǎn)程服務(wù) 
            ChannelFuture future = bootstrap.connect(host, port).sync(); 
            future.channel().closeFuture().sync(); 
        } finally { 
             worker.shutdownGracefully(); 
        } 
    } 
} 

2. 客戶端-(ClientHandler)

package com.haopt.netty.client.handler; 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.channel.SimpleChannelInboundHandler; 
import io.netty.util.CharsetUtil; 
public class MyClientHandler extends SimpleChannelInboundHandler<ByteBuf> { 
    @Override 
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { 
        System.out.println("接收到服務(wù)端的消息：" + 
        msg.toString(CharsetUtil.UTF_8)); 
    } 
    @Override 
    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
        // 向服務(wù)端發(fā)送數(shù)據(jù) 
        String msg = "hello"; 
        ctx.writeAndFlush(Unpooled.copiedBuffer(msg, CharsetUtil.UTF_8)); 
    } 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
        cause.printStackTrace(); 
        ctx.close(); 
    } 
} 

相信代碼執(zhí)行起來(lái)沒(méi)有任何問(wèn)題(如果有任何問(wèn)題反應(yīng)交流)。但是對(duì)上面代碼為何這樣實(shí)現(xiàn)有很多疑。嘿嘿，我也是奧。接下來(lái)我們對(duì)這些代碼中用到的組件進(jìn)行介紹，希望能消除之前疑慮。如果還是不能，可以把疑問(wèn)寫(xiě)于留言處，嘿嘿，我也不一定會(huì)有個(gè)好的解答，但是大佬總會(huì)有的。

02二 Netty核心組件

我們都知道Netty是基于事件驅(qū)動(dòng)。但是事件發(fā)生后，Netty的各個(gè)組件都做了什么?來(lái)看看下面內(nèi)容!

3 2.1 Channel

1. 初識(shí)Channel

a 可以理解為socket連接，客戶端和服務(wù)端連接的時(shí)候會(huì)創(chuàng)建一個(gè)channel。 
負(fù)責(zé)基本的IO操作，例如：bind()、connect()、read()、write()。 
b Netty的Channel接口所提供的API，大大減少了Socket類復(fù)雜性 

2. 常見(jiàn)Channel(不同的協(xié)議和阻塞類型的連接會(huì)有不同的Channel類型與之對(duì)應(yīng))

a NioSocketChannel，NIO的客戶端 TCP Socket 連接。 
 
b NioServerSocketChannel，NIO的服務(wù)器端 TCP Socket 連接。 
 
c NioDatagramChannel， UDP 連接。 
 
d NioSctpChannel，客戶端 Sctp 連接。 
 
e NioSctpServerChannel，Sctp 服務(wù)器端連接，這些通道涵蓋了UDP和TCP⽹絡(luò)IO以及⽂件IO。 

4 2.2 EventLoopGroup、EventLoop

1. 概述

有了Channel連接服務(wù)，連接之間消息流動(dòng)。服務(wù)器發(fā)出消息稱為出站，服務(wù)器接受消息稱為入站。 
那么消息出站和入站就產(chǎn)生了事件例如：連接已激活；數(shù)據(jù)讀取；用戶事件；異常事件；打開(kāi)連接； 
關(guān)閉連接等等。有了事件，有了事件就需要機(jī)制來(lái)監(jiān)控和協(xié)調(diào)事件，這個(gè)機(jī)制就是EventLoop。 

2. 初識(shí)EventLoopGroup、EventLoop

對(duì)上圖解釋

a 一個(gè)EventLoopGroup包含一個(gè)或者多個(gè)EventLoop 
b 一個(gè)EventLoop在生命周期內(nèi)之和一個(gè)Thread綁定 
c EventLoop上所有的IO事件在它專有的Thread上被處理。 
d Channel在它生命周期只注冊(cè)于一個(gè)Event Loop 
e 一個(gè)Event Loop可能被分配給一個(gè)或者多個(gè)Channel 

3. 代碼實(shí)現(xiàn)

// 主線程，不處理任何業(yè)務(wù)邏輯，只是接收客戶的連接請(qǐng)求 
EventLoopGroup boss = new NioEventLoopGroup(1); 
// ⼯作線程，線程數(shù)默認(rèn)是：cpu*2 
EventLoopGroup worker = new NioEventLoopGroup(); 

5 2.3 ChannelHandler

1. 初識(shí)ChannelHandler

對(duì)于數(shù)據(jù)的出站和入棧的業(yè)務(wù)邏輯都是在ChannelHandler中。 

2. 對(duì)于出站和入站對(duì)應(yīng)的ChannelHandler

ChannelInboundHandler ⼊站事件處理器 
ChannelOutBoundHandler 出站事件處理器 

3. 開(kāi)發(fā)中常用的ChannelHandler(ChannelInboundHandlerAdapter、SimpleChannelInboundHandler)

a 源碼

b SimpleChannelInboundHandler的源碼(是ChannelInboundHandlerAdapter子類)

注意：

兩者的區(qū)別在于，前者不會(huì)釋放消息數(shù)據(jù)的引⽤，⽽后者會(huì)釋放消息數(shù)據(jù)的引⽤。

6 2.4 ChannelPipeline 

1. 初識(shí)ChannelPipeline

將ChannelHandler串起來(lái)。一個(gè)Channel包含一個(gè)ChannelPipeline,而ChannelPipeline維護(hù)者一個(gè)ChannelHandler列表。 
ChannelHandler與Channel和ChannelPipeline之間的映射關(guān)系，由ChannelHandlerContext進(jìn)⾏維護(hù)。 

 

如上圖解釋

ChannelHandler按照加⼊的順序會(huì)組成⼀個(gè)雙向鏈表，⼊站事件從鏈表的head往后傳遞到最后⼀個(gè)ChannelHandler。 
出站事件從鏈表的tail向前傳遞，直到最后⼀個(gè)ChannelHandler，兩種類型的ChannelHandler相互不會(huì)影響。 

7 2.5 Bootstrap

1. 初識(shí)Bootstrap

是引導(dǎo)作用，配置整個(gè)netty程序，將各個(gè)組件串起來(lái)，最后綁定接口，啟動(dòng)服務(wù)。 

2. Bootstrap兩種類型(Bootstrap、ServerBootstrap)

客戶端只需要一個(gè)EventLoopGroup，服務(wù)端需要兩個(gè)EventLoopGroup。 

 

上圖解釋

與ServerChannel相關(guān)聯(lián)的EventLoopGroup 將分配⼀個(gè)負(fù)責(zé)為傳⼊連接請(qǐng)求創(chuàng)建 Channel 的EventLoop。 
⼀旦連接被接受，第⼆個(gè) EventLoopGroup 就會(huì)給它的 Channel 分配⼀個(gè) EventLoop。 

8 2.6 Future

1. 初識(shí)

操作完成時(shí)通知應(yīng)用程序的方式。這個(gè)對(duì)象可以看做異步操作執(zhí)行結(jié)果占位符，它在將來(lái)某個(gè)時(shí)刻完成，并提供對(duì)其結(jié)果的訪問(wèn)。 

2. ChannelFuture的由來(lái)

JDK 預(yù)置了 interface java.util.concurrent.Future，但是其所提供的實(shí)現(xiàn)， 
只允許⼿動(dòng)檢查對(duì)應(yīng)的操作是否已經(jīng)完成，或者⼀直阻塞直到它完成。這是⾮常 
繁瑣的，所以 Netty 提供了它⾃⼰的實(shí)現(xiàn)--ChannelFuture，⽤于在執(zhí)⾏異步 
操作的時(shí)候使⽤。 

3. Netty為什么完全是異步?

a ChannelFuture提供了⼏種額外的⽅法，這些⽅法使得我們能夠注冊(cè)⼀個(gè)或者多個(gè)  ChannelFutureListener實(shí)例。 
 
b 監(jiān)聽(tīng)器的回調(diào)⽅法operationComplete()，將會(huì)在對(duì)應(yīng)的操作完成時(shí)被調(diào)⽤。 
 然后監(jiān)聽(tīng)器可以判斷該操作是成功地完成了還是出錯(cuò)了。 
   
c 每個(gè) Netty 的出站 I/O 操作都將返回⼀個(gè) ChannelFuture，也就是說(shuō)， 
 它們都不會(huì)阻塞。所以說(shuō)，Netty完全是異步和事件驅(qū)動(dòng)的。 

9 2.7 組件小結(jié)

上圖解釋

將組件串起來(lái) 

03三 緩存區(qū)-ByteBuf

ByteBuf是我們開(kāi)發(fā)中代碼操作最多部分和出現(xiàn)問(wèn)題最多的一部分。比如常見(jiàn)的TCP協(xié)議通信的粘包和拆包解決，和ByteBuf密切相關(guān)。后面文章會(huì)詳細(xì)分析，先不展開(kāi)。我們這里先了解ByteBuf的常用API和執(zhí)行內(nèi)幕。

10 3.1 ByteBuf概述

1. 初識(shí)ByteBuf

JavaNIO提供了緩存容器(ByteBuffer)，但是使用復(fù)雜。因此netty引入緩存ButeBuf， 
一串字節(jié)數(shù)組構(gòu)成。 

2. ByteBuf兩個(gè)索引(readerIndex，writerIndex)

a readerIndex 將會(huì)根據(jù)讀取的字節(jié)數(shù)遞增 
b writerIndex 也會(huì)根據(jù)寫(xiě)⼊的字節(jié)數(shù)進(jìn)⾏遞增 
 
注意：如果readerIndex超過(guò)了writerIndex的時(shí)候，Netty會(huì)拋出IndexOutOf-BoundsException異常。 

 

11 3.2 ByteBuf基本使用

1. 讀取

package com.haopt.netty.myrpc.test; 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.util.CharsetUtil; 
public class TestByteBuf01 { 
    public static void main(String[] args) { 
        //構(gòu)造 
        ByteBuf byteBuf = Unpooled.copiedBuffer("hello world", 
        CharsetUtil.UTF_8); 
        System.out.println("byteBuf的容量為：" + byteBuf.capacity()); 
        System.out.println("byteBuf的可讀容量為：" + byteBuf.readableBytes()); 
        System.out.println("byteBuf的可寫(xiě)容量為：" + byteBuf.writableBytes()); 
        while (byteBuf.isReadable()){ //⽅法⼀：內(nèi)部通過(guò)移動(dòng)readerIndex進(jìn)⾏讀取 
         System.out.println((char)byteBuf.readByte()); 
        } 
        //⽅法⼆：通過(guò)下標(biāo)直接讀取 
        for (int i = 0; i < byteBuf.readableBytes(); i++) { 
         System.out.println((char)byteBuf.getByte(i)); 
        } 
        //⽅法三：轉(zhuǎn)化為byte[]進(jìn)⾏讀取 
        byte[] bytes = byteBuf.array(); 
        for (byte b : bytes) { 
        System.out.println((char)b); 
        } 
    } 
} 

2. 寫(xiě)入

package com.haopt.netty.myrpc.test; 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.util.CharsetUtil; 
public class TestByteBuf02 { 
    public static void main(String[] args) { 
        //構(gòu)造空的字節(jié)緩沖區(qū)，初始⼤⼩為10，最⼤為20 
        ByteBuf byteBuf = Unpooled.buffer(10,20); 
        System.out.println("byteBuf的容量為：" + byteBuf.capacity()); 
        System.out.println("byteBuf的可讀容量為：" + byteBuf.readableBytes()); 
        System.out.println("byteBuf的可寫(xiě)容量為：" + byteBuf.writableBytes()); 
        for (int i = 0; i < 5; i++) { 
         byteBuf.writeInt(i); //寫(xiě)⼊int類型，⼀個(gè)int占4個(gè)字節(jié) 
        } 
        System.out.println("ok"); 
        System.out.println("byteBuf的容量為：" + byteBuf.capacity()); 
        System.out.println("byteBuf的可讀容量為：" + byteBuf.readableBytes()); 
        System.out.println("byteBuf的可寫(xiě)容量為：" + byteBuf.writableBytes()); 
        while (byteBuf.isReadable()){ 
         System.out.println(byteBuf.readInt()); 
        } 
    } 
} 

3. 丟棄已讀字節(jié)

package com.haopt.netty.myrpc.test; 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.util.CharsetUtil; 
public class TestByteBuf03 { 
    public static void main(String[] args) { 
        ByteBuf byteBuf = Unpooled.copiedBuffer("hello world",CharsetUtil.UTF_8); 
        System.out.println("byteBuf的容量為：" + byteBuf.capacity()); 
        System.out.println("byteBuf的可讀容量為：" + byteBuf.readableBytes()); 
        System.out.println("byteBuf的可寫(xiě)容量為：" + byteBuf.writableBytes()); 
        while (byteBuf.isReadable()){ 
         System.out.println((char)byteBuf.readByte()); 
        } 
        byteBuf.discardReadBytes(); //丟棄已讀的字節(jié)空間 
        System.out.println("byteBuf的容量為：" + byteBuf.capacity()); 
        System.out.println("byteBuf的可讀容量為：" + byteBuf.readableBytes()); 
        System.out.println("byteBuf的可寫(xiě)容量為：" + byteBuf.writableBytes()); 
    } 

4. clear()

package com.haopt.netty.myrpc.test; 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.util.CharsetUtil; 
public class TestByteBuf04 { 
    public static void main(String[] args) { 
        ByteBuf byteBuf = Unpooled.copiedBuffer("hello world",CharsetUtil.UTF_8); 
        System.out.println("byteBuf的容量為：" + byteBuf.capacity()); 
        System.out.println("byteBuf的可讀容量為：" + byteBuf.readableBytes()); 
        System.out.println("byteBuf的可寫(xiě)容量為：" + byteBuf.writableBytes()); 
        byteBuf.clear(); //重置readerIndex 、 writerIndex 為0 
        System.out.println("byteBuf的容量為：" + byteBuf.capacity()); 
        System.out.println("byteBuf的可讀容量為：" + byteBuf.readableBytes()); 
        System.out.println("byteBuf的可寫(xiě)容量為：" + byteBuf.writableBytes()); 
    } 
} 

12 3.3 ByteBuf 使⽤模式

3.3.1 根據(jù)存放緩沖區(qū)，分為三類

1. 堆緩存區(qū)(HeapByteBuf)

內(nèi)存的分配和回收速度⽐較快，可以被JVM⾃動(dòng)回收，缺點(diǎn)是，如果進(jìn)⾏socket的IO讀寫(xiě)，需要額外做⼀次內(nèi)存復(fù)制，將堆內(nèi)存對(duì)應(yīng)的緩沖區(qū)復(fù)制到內(nèi)核Channel中，性能會(huì)有⼀定程度的下降。 
由于在堆上被 JVM 管理，在不被使⽤時(shí)可以快速釋放?？梢酝ㄟ^(guò) ByteBuf.array() 來(lái)獲取 byte[] 數(shù) 
據(jù)。 

2. 直接緩存區(qū)(DirectByteBuf)

⾮堆內(nèi)存，它在對(duì)外進(jìn)⾏內(nèi)存分配，相⽐堆內(nèi)存，它的分配和回收速度會(huì)慢⼀些，但是 
將它寫(xiě)⼊或從Socket Channel中讀取時(shí)，由于減少了⼀次內(nèi)存拷⻉，速度⽐堆內(nèi)存塊。 

3. 復(fù)合緩存區(qū)

顧名思義就是將上述兩類緩沖區(qū)聚合在⼀起。Netty 提供了⼀個(gè) CompsiteByteBuf， 
可以將堆緩沖區(qū)和直接緩沖區(qū)的數(shù)據(jù)放在⼀起，讓使⽤更加⽅便。 

3.3.2 緩存區(qū)選擇

Netty默認(rèn)使⽤的是直接緩沖區(qū)(DirectByteBuf)，如果需要使⽤堆緩沖區(qū)(HeapByteBuf)模式，則需要進(jìn)⾏系統(tǒng)參數(shù)的設(shè)置。

//netty中IO操作都是基于Unsafe完成的 
System.setProperty("io.netty.noUnsafe", "true");  
//ByteBuf的分配要設(shè)置為⾮池化，否則不能切換到堆緩沖器模式 
serverBootstrap.childOption(ChannelOption.ALLOCATOR,UnpooledByteBufAllocator.DEFAULT); 

3.3.3 ByteBuf對(duì)象是否池化(Netty是默認(rèn)池化的)

1. 池化化和非池化的實(shí)現(xiàn)

PooledByteBufAllocator，實(shí)現(xiàn)了ByteBuf的對(duì)象的池化，提⾼性能減少并最⼤限度地減少內(nèi)存碎⽚。 
UnpooledByteBufAllocator，沒(méi)有實(shí)現(xiàn)對(duì)象的池化，每次會(huì)⽣成新的對(duì)象實(shí)例。 

2. 代碼實(shí)現(xiàn)(讓Netty中ByteBuf對(duì)象不池化)

//通過(guò)ChannelHandlerContext獲取ByteBufAllocator實(shí)例 
ctx.alloc(); 
//通過(guò)channel也可以獲取 
channel.alloc(); 
 
//Netty默認(rèn)使⽤了PooledByteBufAllocator 
 
//可以在引導(dǎo)類中設(shè)置⾮池化模式 
serverBootstrap.childOption(ChannelOption.ALLOCATOR,UnpooledByteBufAllocator.DEFAULT); 
//或通過(guò)系統(tǒng)參數(shù)設(shè)置 
System.setProperty("io.netty.allocator.type", "pooled"); 
System.setProperty("io.netty.allocator.type", "unpooled"); 

我在開(kāi)發(fā)項(xiàng)目中，我一般不進(jìn)行更改。因?yàn)槲矣X(jué)得池化效率更高。有其他高見(jiàn)，歡迎留言。

13 3.5 ByteBuf的釋放

ByteBuf如果采⽤的是堆緩沖區(qū)模式的話，可以由GC回收，但是如果采⽤的是直接緩沖區(qū)，就不受GC的 管理，就得⼿動(dòng)釋放，否則會(huì)發(fā)⽣內(nèi)存泄露。

3.5.1 ByteBuf的手動(dòng)釋放(一般不推薦使用，了解)

1. 實(shí)現(xiàn)邏輯

⼿動(dòng)釋放，就是在使⽤完成后，調(diào)⽤ReferenceCountUtil.release(byteBuf); 進(jìn)⾏釋放。 
通過(guò)release⽅法減去 byteBuf的使⽤計(jì)數(shù)，Netty 會(huì)⾃動(dòng)回收 byteBuf。 

2. 代碼

/** 
* 獲取客戶端發(fā)來(lái)的數(shù)據(jù) 
* 
* @param ctx 
* @param msg 
* @throws Exception 
*/ 
@Override 
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
    ByteBuf byteBuf = (ByteBuf) msg; 
    String msgStr = byteBuf.toString(CharsetUtil.UTF_8); 
    System.out.println("客戶端發(fā)來(lái)數(shù)據(jù)：" + msgStr); 
    //釋放資源 
    ReferenceCountUtil.release(byteBuf); 
} 

注意：

⼿動(dòng)釋放可以達(dá)到⽬的，但是這種⽅式會(huì)⽐較繁瑣，如果⼀旦忘記釋放就可能會(huì)造成內(nèi)存泄露。

3.5.1 ByteBuf的自動(dòng)釋放

⾃動(dòng)釋放有三種⽅式，分別是：⼊站的TailHandler、繼承SimpleChannelInboundHandler、 HeadHandler的出站釋放。

1. TailHandler

Netty的ChannelPipleline的流⽔線的末端是TailHandler，默認(rèn)情況下如果每個(gè)⼊站處理器Handler都把消息往下傳，TailHandler會(huì)釋放掉ReferenceCounted類型的消息。

/** 
* 獲取客戶端發(fā)來(lái)的數(shù)據(jù) 
* @param ctx 
* @param msg 
* @throws Exception 
*/ 
@Override 
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
    ByteBuf byteBuf = (ByteBuf) msg; 
    String msgStr = byteBuf.toString(CharsetUtil.UTF_8); 
    System.out.println("客戶端發(fā)來(lái)數(shù)據(jù)：" + msgStr); 
    //向客戶端發(fā)送數(shù)據(jù) 
    ctx.writeAndFlush(Unpooled.copiedBuffer("ok", CharsetUtil.UTF_8)); 
    ctx.fireChannelRead(msg); //將ByteBuf向下傳遞 
} 

源碼：

在DefaultChannelPipeline中的TailContext內(nèi)部類會(huì)在最后執(zhí)⾏

@Override 
public void channelRead(ChannelHandlerContext ctx, Object msg) { 
 onUnhandledInboundMessage(ctx, msg); 
} 
//最后會(huì)執(zhí)⾏ 
protected void onUnhandledInboundMessage(Object msg) { 
  try { 
      logger.debug( 
      "Discarded inbound message {} that reached at the tail of the 
      pipeline. " + "Please check your pipeline configuration.", msg); 
  } finally { 
    ReferenceCountUtil.release(msg); //釋放資源 
  } 
} 

2. SimpleChannelInboundHandler

當(dāng)ChannelHandler繼承了SimpleChannelInboundHandler后，在SimpleChannelInboundHandler的channelRead()⽅法中，將會(huì)進(jìn)⾏資源的釋放。

SimpleChannelInboundHandler的源碼

//SimpleChannelInboundHandler中的channelRead() 
@Override 
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
    boolean release = true; 
    try { 
      if (acceptInboundMessage(msg)) { 
        @SuppressWarnings("unchecked") 
        I imsg = (I) msg; 
        channelRead0(ctx, imsg); 
      } else { 
        release = false; 
        ctx.fireChannelRead(msg); 
      } 
    } finally { 
      if (autoRelease && release) { 
       ReferenceCountUtil.release(msg); //在這⾥釋放 
      } 
    } 
} 

我們handler代碼編寫(xiě)：

package com.haopt.myrpc.client.handler; 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.channel.SimpleChannelInboundHandler; 
import io.netty.util.CharsetUtil; 
public class MyClientHandler extends SimpleChannelInboundHandler<ByteBuf> { 
    @Override 
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { 
      System.out.println("接收到服務(wù)端的消息：" + 
      msg.toString(CharsetUtil.UTF_8)); 
    } 
    @Override 
    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
      // 向服務(wù)端發(fā)送數(shù)據(jù) 
      String msg = "hello"; 
      ctx.writeAndFlush(Unpooled.copiedBuffer(msg, CharsetUtil.UTF_8)); 
    } 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
      cause.printStackTrace(); 
      ctx.close(); 
    } 
} 

3. 堆緩沖區(qū)(HeadHandler)

出站處理流程中，申請(qǐng)分配到的ByteBuf，通過(guò)HeadHandler完成⾃動(dòng)釋放。

在出站流程開(kāi)始的時(shí)候，通過(guò)調(diào)⽤ctx.writeAndFlush(msg)，Bytebuf緩沖區(qū)開(kāi)始進(jìn)⼊出站處理的pipeline流⽔線。

在每⼀個(gè)出站Handler中的處理完成后，最后消息會(huì)來(lái)到出站的最后⼀棒HeadHandler，再經(jīng)過(guò)⼀輪復(fù)雜的調(diào)⽤，在flush完成后終將被release掉。

package com.haopt.myrpc.client.handler; 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.channel.SimpleChannelInboundHandler; 
import io.netty.util.CharsetUtil; 
public class MyClientHandler extends SimpleChannelInboundHandler<ByteBuf> { 
@Override 
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws 
Exception { 
System.out.println("接收到服務(wù)端的消息：" + 
msg.toString(CharsetUtil.UTF_8)); 
} 
@Override 
public void channelActive(ChannelHandlerContext ctx) throws Exception { 
// 向服務(wù)端發(fā)送數(shù)據(jù) 
String msg = "hello"; 
ctx.writeAndFlush(Unpooled.copiedBuffer(msg, CharsetUtil.UTF_8)); 
} 
@Override 
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) 
throws Exception { 
cause.printStackTrace(); 
ctx.close(); 
} 
} 

 

14 3.6 ByteBuf小結(jié)

a ⼊站流程中，如果對(duì)原消息不做處理，調(diào)ctx.fireChannelRead(msg) 把

原消息往下傳，由流⽔線最后⼀棒 TailHandler 完成⾃動(dòng)釋放。

b 如果截?cái)嗔?#12042;站處理流⽔線，則繼承SimpleChannelInboundHandler ，完成⼊站ByteBuf ⾃動(dòng)釋放。

c 出站處理過(guò)程中，申請(qǐng)分配到的 ByteBuf，通過(guò) HeadHandler 完成⾃動(dòng)釋放。

d ⼊站處理中，如果將原消息轉(zhuǎn)化為新的消息ctx.fireChannelRead(newMsg)往下傳，那必須把原消息release掉。

e ⼊站處理中，如果已經(jīng)不再調(diào)⽤ ctx.fireChannelRead(msg) 傳遞任何消息，也沒(méi)有繼承SimpleChannelInboundHandler 完成⾃動(dòng)釋放，那更要把原消息release掉。