一、netty概述

1、NIO存在的問題：

• NIO的API比較復(fù)雜，需要熟練掌握3個(gè)核心組件，channel、buffer和selector；
• 需要熟悉多線程、網(wǎng)絡(luò)編程等技術(shù)；
• 開發(fā)工作量大，難度也比較大，需要解決斷連、重連、網(wǎng)絡(luò)閃斷、半包讀寫、失敗緩存、網(wǎng)絡(luò)擁堵等各種情況；
• NIO存在bug，一個(gè)叫Epoll的bug，會(huì)導(dǎo)致選擇器空輪詢，形成死循環(huán)，最后CPU飆到100%

正是因?yàn)镹IO存在這些問題，netty就應(yīng)運(yùn)而生了。

歡迎大家關(guān)注我的公眾號(hào) javawebkf，目前正在慢慢地將簡書文章搬到公眾號(hào)，以后簡書和公眾號(hào)文章將同步更新，且簡書上的付費(fèi)文章在公眾號(hào)上將免費(fèi)。

2、Netty簡介：
netty是一個(gè)異步的，基于事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用框架。可以快速地開發(fā)高性能的服務(wù)器端和客戶端，像dubbo和elasticsearch底層都用了netty。它具有以下優(yōu)點(diǎn)：

• 設(shè)計(jì)優(yōu)雅，靈活可擴(kuò)展；
• 使用方便，用戶指南清晰明確；
• 安全，完整的SSL/TLS和StartTLS支持；
• 社區(qū)活躍，不斷地更新完善

官方下載地址：https://bintray.com/netty/downloads
我本次下載的版本是：4.1.51.Final

二、netty的架構(gòu)設(shè)計(jì)

1、線程模型：
目前存在的線程模式：

• 傳統(tǒng)阻塞IO的服務(wù)模型
• Reactor模式

根據(jù)Reactor的數(shù)量和1處理資源的線程數(shù)不同，又分3種：

• 單Reactor單線程；
• 單Reactor多線程；
• 主從Reactor多線程

Netty的線程模型是基于主從Reactor多線程做了改進(jìn)。

2、傳統(tǒng)阻塞IO的線程模型：
采用阻塞IO獲取輸入的數(shù)據(jù)，每個(gè)連接都需要獨(dú)立的線程來處理邏輯。存在的問題就是，當(dāng)并發(fā)數(shù)很大時(shí)，就需要?jiǎng)?chuàng)建很多的線程，占用大量的資源。連接創(chuàng)建后，如果當(dāng)前線程沒有數(shù)據(jù)可讀，該線程將會(huì)阻塞在讀數(shù)據(jù)的方法上，造成線程資源浪費(fèi)。

3、Reactor模式(分發(fā)者模式/反應(yīng)器模式/通知者模式)：
針對傳統(tǒng)阻塞IO的模型，做了以下兩點(diǎn)改進(jìn)：

• 基于IO復(fù)用模型：多個(gè)客戶端共用一個(gè)阻塞對象，而不是每個(gè)客戶端都對應(yīng)一個(gè)阻塞對象
• 基于線程池復(fù)用線程資源：使用了線程池，而不是每來一個(gè)客戶端就創(chuàng)建一個(gè)線程

Reactor模式的核心組成：

• Reactor：Reactor就是多個(gè)客戶端共用的那一個(gè)阻塞對象，它單獨(dú)起一個(gè)線程運(yùn)行，負(fù)責(zé)監(jiān)聽和分發(fā)事件，將請求分發(fā)給適當(dāng)?shù)奶幚沓绦騺磉M(jìn)行處理
• Handler：處理程序要完成的實(shí)際事件，也就是真正執(zhí)行業(yè)務(wù)邏輯的程序，它是非阻塞的

4、單Reactor單線程：

模型圖

這個(gè)圖其實(shí)就跟之前的NIO群聊系統(tǒng)對應(yīng)。多個(gè)客戶端請求連接，然后Reactor通過selector輪詢判斷哪些通道是有事件發(fā)生的，如果是連接事件，就到了Acceptor中建立連接；如果是其他讀寫事件，就有dispatch分發(fā)到對應(yīng)的handler中進(jìn)行處理。這種模式的缺點(diǎn)就是Reactor和Handler是在一個(gè)線程中的，如果Handler阻塞了，那么程序就阻塞了。

5、單Reactor多線程：

單reactor多線程

處理流程如下：

• Reactor對象通過Selector監(jiān)聽客戶端請求事件，通過dispatch進(jìn)行分發(fā)；
• 如果是連接事件，則由Acceptor通過accept方法處理連接請求，然后創(chuàng)建一個(gè)Handler對象響應(yīng)事件；
• 如果不是連接請求，則由Reactor對象調(diào)用對應(yīng)handler對象進(jìn)行處理；handler只響應(yīng)事件，不做具體的業(yè)務(wù)處理，它通過read方法讀取數(shù)據(jù)后，會(huì)分發(fā)給線程池的某個(gè)線程進(jìn)行業(yè)務(wù)處理，并將處理結(jié)果返回給handler；
• handler收到響應(yīng)后，通過send方法將結(jié)果返回給client。

相比單Reactor單線程，這里將業(yè)務(wù)處理的事情交給了不同的線程去做，發(fā)揮了多核CPU的性能。但是Reactor只有一個(gè)，所有事件的監(jiān)聽和響應(yīng)，都由一個(gè)Reactor去完成，并發(fā)性還是不好。

6、主從Reactor多線程：

主從reactor多線程

這個(gè)模型相比單reactor多線程的區(qū)別就是：專門搞了一個(gè)MainReactor來處理連接事件，如果不是連接事件，就分發(fā)給SubReactor進(jìn)行處理。圖中這個(gè)SubReactor只有一個(gè)，其實(shí)是可以有多個(gè)的，所以性能就上去了。

• 優(yōu)點(diǎn)：父線程與子線程的交互簡單、職責(zé)明確，父線程負(fù)責(zé)接收連接，子線程負(fù)責(zé)完成后續(xù)的業(yè)務(wù)處理；
• 缺點(diǎn)：編程復(fù)雜度高

7、netty的模型：
netty模型是基于主從Reactor多線程模型設(shè)計(jì)的，其工作流程如下：

• Netty有兩組線程池，一個(gè)Boss Group，它專門負(fù)責(zé)客戶端連接，另一個(gè)Work Group，專門負(fù)責(zé)網(wǎng)絡(luò)讀寫；
• Boss Group和Work Group的類型都是NIOEventLoopGroup；
• NIOEventLoopGroup相當(dāng)于一個(gè)事件循環(huán)組，這個(gè)組包含了多個(gè)事件循環(huán)，每一個(gè)循環(huán)都是NIOEventLoop；
• NIOEventLoop表示一個(gè)不斷循環(huán)執(zhí)行處理任務(wù)的線程，每個(gè)NIOEventLoop都有一個(gè)Selector，用于監(jiān)聽綁定在其上的socket；
• Boss Group下的每個(gè)NIOEventLoop的執(zhí)行步驟有3步：(1). 輪詢accept連接事件；(2). 處理accept事件，與client建立連接，生成一個(gè)NioSocketChannel，并將其注冊到某個(gè)work group下的NioEventLoop的selector上；(3). 處理任務(wù)隊(duì)列的任務(wù)，即runAllTasks；
• 每個(gè)Work Group下的NioEventLoop循環(huán)執(zhí)行以下步驟：(1). 輪詢r(jià)ead、write事件；(2). 處理read、write事件，在對應(yīng)的NioSocketChannel處理；(3). 處理任務(wù)隊(duì)列的任務(wù)，即runAllTasks；
• 每個(gè)Work Group下的NioEventLoop在處理業(yè)務(wù)時(shí)，會(huì)使用pipeline(管道)，pipeline中包含了channel，即通過pipeline可以獲取到對應(yīng)的channel，pipeline中維護(hù)了很多的處理器。

netty模型圖如下，對應(yīng)了上面那段流程：

netty模型圖

三、netty入門實(shí)例

使用netty創(chuàng)建一個(gè)服務(wù)端與客戶端，監(jiān)聽6666端口。

1、服務(wù)端：

• NettyServer：

public class NettyServer {

    public static void main(String[] args) throws Exception {
        // 1. 創(chuàng)建boss group (boss group和work group含有的子線程數(shù)默認(rèn)是cpu數(shù) * 2)
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 2. 創(chuàng)建work group
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            // 3. 創(chuàng)建服務(wù)端啟動(dòng)對象
            ServerBootstrap bootstrap = new ServerBootstrap();
            // 4. 配置啟動(dòng)參數(shù)
            bootstrap.group(bossGroup, workGroup) // 設(shè)置兩個(gè)線程組
                     .channel(NioServerSocketChannel.class) // 使用NioSocketChannel 作為服務(wù)器的通道
                     .option(ChannelOption.SO_BACKLOG, 128) // 設(shè)置線程隊(duì)列等待連接個(gè)數(shù)
                     .childOption(ChannelOption.SO_KEEPALIVE, true) // 設(shè)置保持活動(dòng)連接狀態(tài)
                     .childHandler(new ChannelInitializer<SocketChannel>() { // 創(chuàng)建通道初始化對象
                        // 給pipeline設(shè)置處理器
                        @Override
                        protected void initChannel(SocketChannel sc) throws Exception {
                            // 傳入自定義的handler
                            sc.pipeline().addLast(new NettyServerHandler());
                        } 
                    });
            // 5. 啟動(dòng)服務(wù)器并綁定端口
            ChannelFuture cf = bootstrap.bind(6666).sync();
            // 6. 對關(guān)閉通道進(jìn)行監(jiān)聽
            cf.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
        
    }
}

• NettyServerHandler：

public class NettyServerHandler extends ChannelInboundHandlerAdapter{
    
    // 讀取數(shù)據(jù)
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("接收到客戶端消息：" + buf.toString(CharsetUtil.UTF_8));
    }
    
    // 數(shù)據(jù)讀取完畢后
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello，我是服務(wù)端", CharsetUtil.UTF_8));
    }
    
    // 處理異常
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

2、客戶端：

• NettyClient：

public class NettyClient {

    public static void main(String[] args) throws Exception {
        // 1. 創(chuàng)建事件循環(huán)組
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            // 2. 創(chuàng)建啟動(dòng)對象
            Bootstrap bootstrap = new Bootstrap();
            // 3. 設(shè)置相關(guān)參數(shù)
            bootstrap.group(eventLoopGroup) // 設(shè)置線程組
                     .channel(NioSocketChannel.class) // 設(shè)置通道
                     .handler(new ChannelInitializer<SocketChannel>() {
                        // 給pipeline設(shè)置處理器
                        @Override
                        protected void initChannel(SocketChannel sc) throws Exception {
                            sc.pipeline().addLast(new NettyClientHandler());
                        }
                    });
            // 4. 連接服務(wù)端
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6666).sync();
            // 5. 監(jiān)聽通道關(guān)閉
            channelFuture.channel().closeFuture().sync();
        } finally {
            eventLoopGroup.shutdownGracefully();
        }
        
    }
}

• NettyClientHandler：

public class NettyClientHandler extends ChannelInboundHandlerAdapter{

    // 通道就緒就被觸發(fā)
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client：" + ctx);
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello，我是客戶端", CharsetUtil.UTF_8));
    }
    
    // 讀取數(shù)據(jù)
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("接收到服務(wù)端消息：" + buf.toString(CharsetUtil.UTF_8));
    }
    
    // 處理異常
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

先啟動(dòng)服務(wù)端，然后啟動(dòng)客戶端，就能看到服務(wù)端和客戶端在控制臺(tái)打印出來的消息了。

3、TaskQueue自定義任務(wù)：
上面服務(wù)端的NettyServerHandler的channelRead方法中，假如有一個(gè)非常耗時(shí)的業(yè)務(wù)，那么就會(huì)阻塞在那里，直到業(yè)務(wù)執(zhí)行完。比如將NettyServerHandler的channelRead方法改成下面這樣：

// 讀取數(shù)據(jù)
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    // 線程休眠10秒，模擬耗時(shí)業(yè)務(wù)
    TimeUnit.SECONDS.sleep(10);
    ByteBuf buf = (ByteBuf) msg;
    System.out.println("接收到客戶端消息：" + buf.toString(CharsetUtil.UTF_8));
}

啟動(dòng)后會(huì)發(fā)現(xiàn)，10秒鐘后，服務(wù)端才會(huì)收到客戶端發(fā)送的消息，客戶端也要10秒后，才會(huì)接收到服務(wù)端的消息，即服務(wù)端的channelReadComplete方法是要在channelRead方法執(zhí)行完后才會(huì)執(zhí)行的。

一直阻塞在那里也不是辦法，希望可以異步執(zhí)行，那我們就可以把該任務(wù)提交到該channel對應(yīng)的NioEventLoop的TaskQueue中。有以下解決方案：

• 用戶程序自定義的普通任務(wù)：將channelRead方法改成下面這樣：

// 讀取數(shù)據(jù)
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ctx.channel().eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) { e.printStackTrace();}
            ByteBuf buf = (ByteBuf) msg;
            System.out.println("接收到客戶端消息：" + buf.toString(CharsetUtil.UTF_8));
        }
    });
}

這里仍舊休眠10秒。啟動(dòng)服務(wù)端，再啟動(dòng)客戶端，發(fā)現(xiàn)服務(wù)端要10s后才會(huì)打印出客戶端發(fā)送的消息，但是客戶端立刻就可以收到服務(wù)端在channelReadComplete方法里發(fā)送的消息，說明這次是異步的。

• 用戶自定義定時(shí)任務(wù)：與上面的區(qū)別不大，代碼如下：

// 讀取數(shù)據(jù)
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ctx.channel().eventLoop().schedule(new Runnable() {
        @Override
        public void run() {
            try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) { e.printStackTrace();}
            ByteBuf buf = (ByteBuf) msg;
            System.out.println("接收到客戶端消息：" + buf.toString(CharsetUtil.UTF_8));
        }
    }, 5, TimeUnit.SECONDS);
}

啟動(dòng)服務(wù)端，然后啟動(dòng)客戶端，發(fā)現(xiàn)客戶端還是會(huì)立即收到服務(wù)端發(fā)出的消息，而服務(wù)端，首先要等待5秒才去執(zhí)行run方法，run方法里面線程又休眠了10秒，所以總共要15秒后才會(huì)打印出客戶端發(fā)送的消息。

• 非當(dāng)前Reactor線程調(diào)用channel的各種方法：這個(gè)的意思就是，如果我別的業(yè)務(wù)代碼，比如消息推送系統(tǒng)，也想給客戶端發(fā)送消息，該咋整？其實(shí)很簡單，在NettyServerHandler的channelRead方法里，我們是通過ChannelHandlerContext 拿到Channel然后進(jìn)行各種操作的，所以拿到了Channel就可以進(jìn)行操作。那么可以在NettyServer中將Channel保存到集合中去，然后遍歷集合，拿到Channel就可以進(jìn)行操作了。

// 給pipeline設(shè)置處理器
@Override
protected void initChannel(SocketChannel sc) throws Exception {
    // 傳入自定義的handler
    sc.pipeline().addLast(new NettyServerHandler());
    // 在這里，可以將SocketChannel sc保存到集合中，別的線程拿到集合就可以調(diào)用channel的方法了
} 

四、Netty的異步模型

1、基本介紹：

• Netty中的I/O操作都是異步的，包括bind、write和connect。這些操作會(huì)返回一個(gè)ChannelFuture對象，而不會(huì)立即返回操作結(jié)果。
• 調(diào)用者不能立即得到返回結(jié)果，而是通過Futrue-Listener機(jī)制，用戶可以主動(dòng)獲取或者通過通知機(jī)制獲得IO操作的結(jié)果。
• Netty的異步是建立在future和callback之上的。callback是回調(diào)，future表示異步執(zhí)行的結(jié)果，它的核心思想是：假設(shè)有個(gè)方法fun()，計(jì)算過程可能非常耗時(shí)，等待fun()返回要很久，那么可以在調(diào)用fun()的時(shí)候，立馬返回一個(gè)future，后續(xù)通過future去監(jiān)控fun()方法的處理過程，這就是future-listener機(jī)制。
• 用戶可以通過注冊監(jiān)聽函數(shù)，來獲取操作真正的結(jié)果，ChannelFuture常用的函數(shù)如下：

// 判斷當(dāng)前操作是否完成
isDone
// 判斷當(dāng)前操作是否成功
isSuccess
// 獲取操作失敗的原因
getCause
// 判斷當(dāng)前操作是否被取消
isCancelled
// 注冊監(jiān)聽器
addListener

2、使用監(jiān)聽器：
在NettyServer中的“啟動(dòng)并綁定端口”下面加上如下代碼：

// 5. 啟動(dòng)服務(wù)器并綁定端口
ChannelFuture cf = bootstrap.bind(6666).sync();
// 注冊監(jiān)聽器
cf.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture cf) throws Exception {
        if (cf.isSuccess()) {
            System.out.println("綁定端口成功");
        } else {
            System.out.println("綁定端口失敗");
        }
    }
});

這樣就注冊了監(jiān)聽器，會(huì)監(jiān)聽綁定端口的結(jié)果，如果成功了，就會(huì)打印出綁定成功這段話。

五、使用Netty開發(fā)Http服務(wù)

開發(fā)一個(gè)Netty服務(wù)端，監(jiān)聽80端口，瀏覽器訪問localhost，可以返回信息給瀏覽器。代碼如下：

• NettyHttpServer：

public class NettyHttpServer {

    public static void main(String[] args) throws Exception {
        // 1. 創(chuàng)建boss group (boss group和work group含有的子線程數(shù)默認(rèn)是cpu數(shù) * 2)
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 2. 創(chuàng)建work group
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            // 3. 創(chuàng)建服務(wù)端啟動(dòng)對象
            ServerBootstrap bootstrap = new ServerBootstrap();
            // 4. 配置啟動(dòng)參數(shù)
            bootstrap.group(bossGroup, workGroup) // 設(shè)置兩個(gè)線程組
                    .channel(NioServerSocketChannel.class) // 使用NioSocketChannel 作為服務(wù)器的通道
                    .childHandler(new NettyHttpServerInitializer());
            // 5. 啟動(dòng)服務(wù)器并綁定端口
            ChannelFuture cf = bootstrap.bind(80).sync();
            // 6. 對關(guān)閉通道進(jìn)行監(jiān)聽
            cf.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }

}

• NettyHttpServerInitializer：

public class NettyHttpServerInitializer extends ChannelInitializer<SocketChannel> {

    // 向管道加入處理器
    @Override
    protected void initChannel(SocketChannel sc) throws Exception {
        // 1. 得到管道
        ChannelPipeline pipeline = sc.pipeline();
        // 2. 加入一個(gè)編碼器解碼器
        pipeline.addLast("codec", new HttpServerCodec());
        // 3. 增加一個(gè)自定義的handler處理器
        pipeline.addLast("handler", new NettyHttpServerHandler());
    }
}

• NettyHttpServerHandler：

public class NettyHttpServerHandler extends SimpleChannelInboundHandler<HttpObject>{

    // 讀取數(shù)據(jù)
    @Override
    protected void channelRead0(ChannelHandlerContext chc, HttpObject msg) throws Exception {
        // 1. 判斷msg是不是httpRequest請求
        if (msg instanceof HttpRequest) {
            System.out.println("msg類型：" + msg.getClass());
            System.out.println("客戶端地址：" + chc.channel().remoteAddress());
            // 對特定資源不做響應(yīng)
            HttpRequest httpRequest = (HttpRequest) msg;
            URI uri = new URI(httpRequest.uri());
            if ("/favicon.ico".equals(uri.getPath())) {
                System.out.println("請求了圖標(biāo)，不做響應(yīng)");
                return;
            }
            // 2. 創(chuàng)建回復(fù)給瀏覽器的信息
            ByteBuf content = Unpooled.copiedBuffer("hello, 我是服務(wù)器，are you ok?", CharsetUtil.UTF_8);
            // 3. 構(gòu)造http響應(yīng)
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content);
            response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain");
            response.headers().set(HttpHeaderNames.CONTENT_LENGTH, content.readableBytes());
            // 4. 將response返回
            chc.writeAndFlush(response);
        }
    }
}

啟動(dòng)server后，在瀏覽器就訪問localhost就可以返回相關(guān)內(nèi)容了。