摘要：上一篇文章偽異步和我們使用的原生類庫進行異步的開發現在我們使用來進行開發示例代碼綁定端口同步等待成功等待服務端監聽端口關閉創建兩個實例是個線程組它包含了一組線程專門用于網絡事件的處理實際上它們就是線程組這里創建兩個的原因是一個用于服務端接

上一篇文章 BIO、偽異步 IO、AIO和NIO 我們使用 JDK 的 NIO 原生類庫進行異步 IO 的開發. 現在我們使用 Netty 來進行開發.

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();

            // 等待服務端監聽端口關閉
            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();
        }
    }

}

創建兩個 NioEventLoopGroup 實例. NioEventLoopGroup 是個線程組, 它包含了一組 NIO 線程, 專門用于網絡事件的處理, 實際上它們就是 Reactor 線程組. 這里創建兩個的原因是一個用于服務端接收客戶端的連接, 另一個用于進行 SocketChannel 的網絡讀寫.

創建 ServerBootstrap 對象, 它是 Netty 用于啟動 NIO 服務端的輔助啟動類, 目的是降低服務端的開發復雜度.

調用 ServerBootstrap 的 group 方法, 將兩個 NIO 線程組當作參數傳遞到 ServerBootstrap 中. 并且設置 Channel 為 NioServerSocketChannel, 它的功能對于 JDK NIO 類庫中的 ServerSocketChannel 類. 然后配置 NioServerSocketChannel 的 TCP 參數, 此處將它的 backlog 設置為 1024. 最后綁定 IO 事件的處理類 ChildChannelHandler, 它的作用類似于 Reactor 模式中的 Handler 類, 主要用于處理網絡 IO 事件, 例如記錄日志, 對消息進行編解碼等.

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

服務端啟動輔助類配置完成之后, 調用它的 bind 方法綁定監聽端口, 隨后, 調用同步阻塞方法 sync 等待綁定操作完成. 完成之后 Netty 會返回一個 ChannelFuture, 它的功能類似于 JDK 的 java.util.concurrent.Future, 主要用于異步操作的通知回調.

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

調用 NIO 線程組的 shutdownGracefully 進行優雅退出, 它會釋放跟 shutdownGracefully 相關聯的資源.

TimeServerHandler 繼承自 ChannelHandlerAdapter, 它用于對網絡事件進行讀寫操作, 通常我們只需要關注 channelRead 和 exceptionCaught 方法.

在 channelRead 方法中, 將 msg 轉換成 Netty 的 ByteBuf 對象. ByteBuf 類似與 JDK 中的 java.nio.ByteBuffer 對象, 不過它提供了更加強大和靈活的功能. 通過 ByteBuf 的 readableBytes 方法可以獲取緩沖區可讀的字節數. 然后使用 readBytes 方法將緩沖區中的字節數組復制到新建的 byte 數組中.

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

注意: write 方法只是將要發送的消息放到消息發送隊列中, 再通過調用 flush 方法, 將發送緩沖區中的消息全部寫到 SocketChannel 中.

在 exceptionCaught 方法中, 如果出現異常, 就關閉 ChannelHandlerContext, 釋放和 ChannelHandlerContext 相關聯的句柄等資源.

EventLoopGroup: 線程組, 專門用于網絡事件的處理, 實際上它們就是 Reactor 線程組.
ServerBootstrap 是 Netty 用于啟動 NIO 服務端的輔助啟動類.
group方法: 第一個參數 parent (acceptor), 第二個參數 child (client).
NioServerSocketChannel: 它的功能對于 JDK NIO 類庫中的 ServerSocketChannel 類.
ServerSocketChannel: 允許我們監聽TCP鏈接請求, 每個請求會創建會一個 SocketChannel.
ChannelHandlerAdapter類: 它用于對網絡事件進行讀寫操作.