摘要：上一篇文章偽異步和我們使用的原生類庫進(jìn)行異步的開發(fā)現(xiàn)在我們使用來進(jìn)行開發(fā)示例代碼綁定端口同步等待成功等待服務(wù)端監(jiān)聽端口關(guān)閉創(chuàng)建兩個(gè)實(shí)例是個(gè)線程組它包含了一組線程專門用于網(wǎng)絡(luò)事件的處理實(shí)際上它們就是線程組這里創(chuàng)建兩個(gè)的原因是一個(gè)用于服務(wù)端接

上一篇文章 BIO、偽異步 IO、AIO和NIO 我們使用 JDK 的 NIO 原生類庫進(jìn)行異步 IO 的開發(fā). 現(xiàn)在我們使用 Netty 來進(jìn)行開發(fā).

public class TimeServer {

    public void bind(int port) throws Exception {

        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(new ChildChannelHandler());

            // 綁定端口, 同步等待成功
            ChannelFuture f = b.bind(port).sync();

            // 等待服務(wù)端監(jiān)聽端口關(guān)閉
            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    private class ChildChannelHandler extends ChannelInitializer {

        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new TimeServerHandler());
        }
    }

    private class TimeServerHandler extends ChannelHandlerAdapter {
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf buffer = (ByteBuf) msg;
            byte[] req = new byte[buffer.readableBytes()];
            buffer.readBytes(req);

            String body = new String(req, "UTF-8");
            System.out.println(body);

            ByteBuf resp = Unpooled.copiedBuffer("6666".getBytes());
            ctx.write(resp);
        }

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    }

}

創(chuàng)建兩個(gè) NioEventLoopGroup 實(shí)例. NioEventLoopGroup 是個(gè)線程組, 它包含了一組 NIO 線程, 專門用于網(wǎng)絡(luò)事件的處理, 實(shí)際上它們就是 Reactor 線程組. 這里創(chuàng)建兩個(gè)的原因是一個(gè)用于服務(wù)端接收客戶端的連接, 另一個(gè)用于進(jìn)行 SocketChannel 的網(wǎng)絡(luò)讀寫.

創(chuàng)建 ServerBootstrap 對(duì)象, 它是 Netty 用于啟動(dòng) NIO 服務(wù)端的輔助啟動(dòng)類, 目的是降低服務(wù)端的開發(fā)復(fù)雜度.

調(diào)用 ServerBootstrap 的 group 方法, 將兩個(gè) NIO 線程組當(dāng)作參數(shù)傳遞到 ServerBootstrap 中. 并且設(shè)置 Channel 為 NioServerSocketChannel, 它的功能對(duì)于 JDK NIO 類庫中的 ServerSocketChannel 類. 然后配置 NioServerSocketChannel 的 TCP 參數(shù), 此處將它的 backlog 設(shè)置為 1024. 最后綁定 IO 事件的處理類 ChildChannelHandler, 它的作用類似于 Reactor 模式中的 Handler 類, 主要用于處理網(wǎng)絡(luò) IO 事件, 例如記錄日志, 對(duì)消息進(jìn)行編解碼等.

ChannelOption.SO_BACKLOG 對(duì)應(yīng)的是 tcp/ip 協(xié)議 listen 函數(shù)中的 backlog 參數(shù), 函數(shù) listen(int socketfd,int backlog) 用來初始化服務(wù)端可連接隊(duì)列, 服務(wù)端處理客戶端連接請(qǐng)求是順序處理的, 所以同一時(shí)間只能處理一個(gè)客戶端連接, 多個(gè)客戶端來的時(shí)候, 服務(wù)端將不能處理的客戶端連接請(qǐng)求放在隊(duì)列中等待處理, backlog參數(shù)指定了隊(duì)列的大小. ChannelOption 連接

服務(wù)端啟動(dòng)輔助類配置完成之后, 調(diào)用它的 bind 方法綁定監(jiān)聽端口, 隨后, 調(diào)用同步阻塞方法 sync 等待綁定操作完成. 完成之后 Netty 會(huì)返回一個(gè) ChannelFuture, 它的功能類似于 JDK 的 java.util.concurrent.Future, 主要用于異步操作的通知回調(diào).

f.channel().closeFuture().sync(); 方法進(jìn)行阻塞, 等待服務(wù)端鏈路關(guān)閉之后 main 函數(shù)才退出.

調(diào)用 NIO 線程組的 shutdownGracefully 進(jìn)行優(yōu)雅退出, 它會(huì)釋放跟 shutdownGracefully 相關(guān)聯(lián)的資源.

TimeServerHandler 繼承自 ChannelHandlerAdapter, 它用于對(duì)網(wǎng)絡(luò)事件進(jìn)行讀寫操作, 通常我們只需要關(guān)注 channelRead 和 exceptionCaught 方法.

在 channelRead 方法中, 將 msg 轉(zhuǎn)換成 Netty 的 ByteBuf 對(duì)象. ByteBuf 類似與 JDK 中的 java.nio.ByteBuffer 對(duì)象, 不過它提供了更加強(qiáng)大和靈活的功能. 通過 ByteBuf 的 readableBytes 方法可以獲取緩沖區(qū)可讀的字節(jié)數(shù). 然后使用 readBytes 方法將緩沖區(qū)中的字節(jié)數(shù)組復(fù)制到新建的 byte 數(shù)組中.

我們還調(diào)用了 ChannelHandlerContext 的 flush 方法, 它的作用是將消息發(fā)送隊(duì)列中的消息寫入到 SocketChannel 中發(fā)送給對(duì)方. 從性能角度考慮, 為了防止頻繁的喚醒 Selector 進(jìn)行消息發(fā)送, Netty 的 write 方法并不直接將消息寫入 SocketChannel 中.

注意: write 方法只是將要發(fā)送的消息放到消息發(fā)送隊(duì)列中, 再通過調(diào)用 flush 方法, 將發(fā)送緩沖區(qū)中的消息全部寫到 SocketChannel 中.

在 exceptionCaught 方法中, 如果出現(xiàn)異常, 就關(guān)閉 ChannelHandlerContext, 釋放和 ChannelHandlerContext 相關(guān)聯(lián)的句柄等資源.

EventLoopGroup: 線程組, 專門用于網(wǎng)絡(luò)事件的處理, 實(shí)際上它們就是 Reactor 線程組.
ServerBootstrap 是 Netty 用于啟動(dòng) NIO 服務(wù)端的輔助啟動(dòng)類.
group方法: 第一個(gè)參數(shù) parent (acceptor), 第二個(gè)參數(shù) child (client).
NioServerSocketChannel: 它的功能對(duì)于 JDK NIO 類庫中的 ServerSocketChannel 類.
ServerSocketChannel: 允許我們監(jiān)聽TCP鏈接請(qǐng)求, 每個(gè)請(qǐng)求會(huì)創(chuàng)建會(huì)一個(gè) SocketChannel.
ChannelHandlerAdapter類: 它用于對(duì)網(wǎng)絡(luò)事件進(jìn)行讀寫操作.