Netty 框架學習 —— 引導

概述

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

Bootstrap 類

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

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

引導客戶端

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

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 類的方法

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 = https://www.cnblogs.com/Yee-Q/p/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();

標籤：Java

上一篇：Java通用樹結構資料管理

下一篇：最好的Kubernetes客戶端Java庫fabric8io，快來自定義你的操作