Netty 框架學習 —— 引導

概述

前面我們學習了 ChannelPipeline、ChannelHandler 和 EventLoop 之后，接下來的問題是：如何將它們組織起來，成為一個可實際運行的應用程序呢？答案是使用引導（Bootstrap），引導一個應用程序是指對它進行配置，並使它運行起來的過程，也即是將所有的框架組件在后台組合起來並啟用

Bootstrap 類

引導類的層次結構包含一個抽象父類和兩個具體子類

如果將客戶端和服務端視為兩個應用程序，那么它們的功能是不一樣的：服務端致力於使用一個父 Channel 來接受客戶端的連接，並創建子 Channel 以用於它們之間的通信；而客戶端很可能只需要一個單獨的 Channel 來用於所有的網絡交互。這兩種方式之間通用的引導步驟由 AbstractBootstrap 處理，而特定於客戶端或者服務端的引導步驟分別由 Bootstrap 或 ServerBootstrap 處理

引導客戶端

Bootstrap 類被用於客戶端或者使用了無連接協議的應用程序，該類的 API 如表所示：

名稱	描述
Bootstrap group(EventLoopGroup)	設置用於處理 Channel所有事件的 EventLoopGroup
Bootstrap channel(Class<? extends C>) Bootstrap channelFactory(ChannelFactory<? extends C>)	channel() 方法指定了 Channel 的實現類。如果該實現類沒提供默認的構造函數，可以通過調用 channelFactory() 方法來指定一個工廠類，它將會被 bind() 方法調用
Bootstrap localAddress(SocketAddress)	指定 Channel 應該綁定的本地地址，如果沒有指定，則由操作系統創建一個隨機的地址
<T> Bootstrap option(ChannelOption<T> option, T value)	設置 ChannelOption，其將被應用到每個新創建的 Channel 的 ChannelConfig
<T> Bootstrap attr(Attribute<T> key, T value)	指定新創建的 Channel 的屬性值
Bootstrap handler(ChannelHandler)	設置將被添加到 ChannelPipeline 以接收事件通知的 ChannelHandler
Bootstrap remoteAddress(SockerAddress)	設置遠程地址
ChannelFuture connect()	連接到遠程節點並返回一個 ChannelFuture
ChannelFuture bind()	綁定 Channel 並返回一個 ChannelFuture

Bootstrap 類負責為客戶端和使用無連接協議的應用程序創建 Channel

代碼清單展示了引導一個使用 NIO TCP 傳輸的客戶端

EventLoopGroup group = new NioEventLoopGroup();
// 創建一個 Bootstrap 類的實例以創建和連接新的客戶端
Bootstrap bootstrap = new Bootstrap();
// 設置 EventLoopGroup
bootstrap.group(group)
    // 指定要使用的 Channel 實現
    .channel(NioSocketChannel.class)
    // 設置用於 Channel 事件和數據的 ChannelInboundHandler
    .handler(new SimpleChannelInboundHandler<ByteBuf>() {
        @Override
        protected void channeRead0(
        	ChannelHandlerContext channelHandlerContext,
            ByteBuf byteBuf) throws Exception {
            Syetem.out.println("Received data");
        }
    });
// 連接到遠程主機
ChannelFuture future = bootstrap.connect(
	new InetSocketAddress("www.manning.com", 80)	
);
future.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        if(channelFuture.isSuccess()) {
            System.out.println("Connection established");
        } else {
            System.err.println("Connection attempt failed");
            channelFuture.cause().printStackTrace();
        }
    }
})

引導服務端

下表列出了 ServerBootstrap 類的方法

名稱	描述
group	設置 ServerBootstrap 要用的 EventLoopGroup
channel	設置將要被實例化的 ServerChannel 類
channelFactory	如果不能通過默認的構造函數創建 Channel，那么可以提供一個 ChannelFactory
localAddress	指定 ServerChannel 應該綁定的本地地址，如果沒有指定，則由操作系統使用一個隨機地址
option	指定要應用到新創建的 ServerChannel 的 ChannelConfig 的 ChannelOption
childOption	指定當子 Channel 被接受時，應用到子 Channel 的 ChannelConfig 的 ChannelOption
attr	指定 ServerChannel 上的屬性
childAttr	將屬性設置給已經被接受的子 Channel
handler	設置被添加到 ServerChannel 的 ChannelPipeline 中的 ChannelHandler
childHandler	設置將被添加到已被接受的子 Channel 的 ChannelPipeline 中的 ChannelHandler
	綁定 ServerChannel 並且返回一個 ChannelFuture，其將會在綁定操作完成后收到通知

ServerChannel 的實現負責創建子 Channel，這些子 Channel 代表了已被接受的連接。ServerBootstrap 提供了 childHandler()、childAttr() 和 childOption() 這些方法，以簡化將設置應用到已被接受的子 Channel 的 ChannelConfig 的任務

下圖展示了 ServerBootstrap 在 bind() 方法被調用時創建了一個 ServerChannel，並且該 ServerChannel 管理了多個子 Channel

引導服務器的代碼如下所示：

NioEventLoopGroup group = new NioEventLoopGroup();
// 創建 ServerBootstrap
ServerBootstrap bootstrap = new ServerBootstrap();
// 設置 EventLoopGroup
bootstrap.group(group)
    // 指定要使用的 Channel 實現
    .channel(NioServerSocketChannel.class)
    // 設置用於處理已被接受的子 Channel 的 IO 及數據的 ChannelInboundHandler
    .childHandler(new SimpleChannelInboundHandler<ByteBuf>() {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx,
                                   ByteBuf byteBuf) throws Exception {
            System.out.println("Received data");
        }
    });
ChannelFuture future = bootstrap.bind(new InetSocketAddress(8080));
future.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        if(channelFuture.isSuccess()) {
            System.out.println("Server bound");
        } else {
            System.out.println("Bound attempt failed");
            channelFuture.cause().printStackTrace();
        }
    }
})

從 Channel 引導客戶端

假設要求你的服務器充當第三方的客戶端，在這種情況下，需要從已經被接受的子 Channel 中引導一個客戶端 Channel

我們可以按照前面講過的引導客戶端的方式創建新的 Bootstrap 實例，但這要求你為每個新創建的客戶端 Channel 定義一個 EventLoop，這會產生額外的線程，並且子 Channel 和客戶端 Channel 之間交換數據時不可避免會發生上下文切換

一個更好的解決辦法是：通過將子 Channel 的 EventLoop 傳遞給 Bootstrap 的 group() 方法來共享該 EventLoop 傳遞給 Bootstrap 的 group() 方法來共享該 EventLoop，避免額外的線程創建和上下文切換

實現 EventLoop 共享涉及通過調用 group() 方法來設置 EventLoop

ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(new NioEventLoopGroup(), new NioEventLoopGroup())
    .channel(NioServerSocketChannel.class)
    .childHandler(
		new SimpleChannelInboundHandler<ByteBuf>() {
            ChannelFuture connectFuture;
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                // 創建一個 Bootstrap 實例以連接到遠程主機
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.channel(NioSockerChannel.class).handler(
                	new SimpleChannelInboundHandler<ByteBuf>() {
                        @Override
                        protected void channelRead0(
                        	ChannelHandlerContext ctx, ByteBuf in) throws Exception {
                            System.out.println("Received data");
                        }
                    });
                // 使用子 Channel 的 EventLoop
                bootstrap.group(ctx.channel().eventLoop());
                connectFuture = bootstrap.connect(new InetSocketAddress("www.manning.com", 80));
            }
            @Override
            protected void channelRead0(
            	ChannelHandlerContext ctx, ByteBuf byteBuf) throws Exception {
                if(connectFuture.isDone) {
                    // 當連接完成時，執行數據操作
                }
            }
        });

引導過程中添加多個 ChannelHandler

前面的引導過程中調用了 handler() 或者 childHandler() 方法來添加單個 channelHandler() 方法來添加單個 ChannelHandler，如果我們需要多個 ChannelHandler，Netty 提供了一個特殊的 ChannelInboundHandlerAdapter 子類：

public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter

它定義了如下方法

protected abstract void initChannel(C ch) throws Exception;

這個方法提供了一種將多個 ChannelHandler 添加到一個 ChannelPipeline 中的簡便方法，你只需要向 Bootstrap 或 ServerBootstrap 的實例提供你的 ChannelInitializer 實現即可。一旦 Channel 被注冊到它的 EventLoop 之后，就會調用你的 initChannel() 版本，在該方法返回之后，ChannelInitializer 的實例將會從 ChannelPipeline 中移除自己

ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(new NioEventLoopGroup(), new NioEventLoopGroup())
    .channel(NioServerSocketChannel.class)
    // 注冊一個 ChannelInitializerImpl 的實例來設置 ChannelPipeline
    .childHandler(new ChannelInitializerImpl());
ChannelFuture future = bootstrap.bind(new InetSocketAddress(8080));
future.sync();

final class ChannelInitializerImpl extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel ch) throws Exception {
        CHannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new HttpClientCodec());
        pipeline.addLast(new HttpObjectAggregator(Integer.MAX_VALUE));
    }
}

使用 Netty 的 ChannelOption 和屬性

在每個 Channel 創建時都手動配置可能會相當乏味，可以使用 option() 方法來將 ChannelOption 應用到引導，其值會自動應用到所創建的所有 Channel。可用的 ChannelOption 包括了底層連接的詳細信息，如 keep-alive 或者超時屬性以及緩沖區設置

Netty 應用程序通常與組織的專有軟件集成在一起，而 Channel 甚至可能會在正常的 Netty 生命周期之外被使用。在某些常用屬性和數據不可用時，Netty 提供了 AttributeMap 抽象以及 AttributeKey<T>，使用這些工具，可以安全地將任何類型的數據與客戶端和服務端 Channel 相關聯

例如，考慮一個用於跟蹤用戶和 Channel 之間關系的服務器應用程序，可以通過將用戶的 ID 存儲為 Channel 的一個屬性來完成

// 創建一個 AttributeKey 以標識該屬性
final AttributeKey<Integer> id = AttributeKey.newInstance("ID");
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(new NioEventLoopGroup())
    .channel(NioSocketChannel.class)
    .handler(
		new SimpleChannelInboundHandler<ByteBuf>() {
        	@Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                // 使用 AttributeKey 檢索屬性以及它的值
                Integer idValue = ctx.channel().attr(id).get();
            }
            
            @Override
            public void channelRead0(ChannelHandlerContext ctx, ByteBuf byteBuf) throws Exception {
                System.out.println("Received data");
            }
        });
// 設置 ChannelOption
bootstrap.option(ChannelOption.SO_KEEPALIVE, true)
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);
// 存儲 id 屬性
bootstrap.attr(id, 123456);
ChannelFuture future = bootstrap.connect(new InetSocketAddress("www.maning.com", 80));
future.syncUninterruptibly();

引導 DatagramChannel

前面使用的都是基於 TCP 協議的 SocketChannel，但 Bootstrap 類也可以用於無連接協議。為此，Netty 提供了各種 DatagramChannel 的實現，唯一的區別就是，不再調用 connect() 方法，而只調用 bind() 方法

Bootstrap bootstrap = new Bootstrap();
bootstrap.group(new OioEventLoopGroup())
    .channel(OioSocketChannel.class)
    .handler(
		new SimpleChannelInboundHandler<DatagramPacket>() {
            @Override
            public void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
                System.out.println("Received data");
            }
        });
ChannelFuture future = bootstrap.bind(new InetSocketAddress(0));

關閉

引導使得你的應用程序啟動，自然也需要優雅地進行關閉，你也可以讓 JVM 在退出時處理一切，但這不符合優雅的定義

最重要的是，你需要關閉 EventLoopGroup，它將處理任何掛起的事件和任務，並隨后釋放所有活動線程。通過調用 EventLoopGroup.shutdownGracefully() 方法，將返回一個 Future，這個 Future 將在關閉完成時接收到通知。shutdownGracefully 是一個異步操作，你需要阻塞等待直到它完成，或者向返回的 Future 注冊一個監聽器

EventLoopGroup group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
    .channel(NioServerSocketChannel.class);
...
Future<?> future = group.shutdownGracefully();
future.syncUniterruptibly();