netty服務(wù)端輔助類ServerBootstrap創(chuàng)建邏輯分析
ServerBootstrap創(chuàng)建
ServerBootstrap 為 netty 建立服務(wù)端的輔助類, 以 NIO為例,創(chuàng)建代碼如下:
public static void main(String[] args) throws Exception { ServerBootstrap bs = new ServerBootstrap(); bs.group(new NioEventLoopGroup(1), new NioEventLoopGroup()) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<Channel>() { @Override protected void initChannel(Channel ch) throws Exception { ch.pipeline() .addLast(new HttpServerCodec()) .addLast(new HttpObjectAggregator(65535)) .addLast(new Controller()); } }).bind(8080).sync().channel().closeFuture().sync(); }
核心參數(shù)
//配置屬性,如 SO_KEEPALIVE 等private final ServerBootstrapConfig config = new ServerBootstrapConfig(this); //acceot 的 子channel所綁定的 事件循環(huán)組" private volatile EventLoopGroup childGroup; private volatile ChannelHandler childHandler;
初始化流程
主要為 綁定本地端口 -> 注冊自身到 EventLoop , 并注冊 accept 和 read 事件 -> EventLoop的主循環(huán)中會不斷的select注冊的channel的事件,并處理。
首先執(zhí)行綁定
核心邏輯位于
io.netty.bootstrap.AbstractBootstrap.doBind(SocketAddress) 和 io.netty.bootstrap.AbstractBootstrap.initAndRegister()中
private ChannelFuture doBind(final SocketAddress localAddress) { final ChannelFuture regFuture = initAndRegister(); ..........if (regFuture.isDone()) { // At this point we know that the registration was complete and successful. ChannelPromise promise = channel.newPromise(); //綁定邏輯 doBind0(regFuture, channel, localAddress, promise); return promise; } else { // Registration future is almost always fulfilled already, but just in case it's not. final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel); regFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { Throwable cause = future.cause(); if (cause != null) { // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an // IllegalStateException once we try to access the EventLoop of the Channel. promise.setFailure(cause); } else { // Registration was successful, so set the correct executor to use. // See https://github.com/netty/netty/issues/2586 promise.registered(); doBind0(regFuture, channel, localAddress, promise); } } }); return promise; } }
注冊自身到 EventLoop
先來看 initAndRegister , 核心邏輯就是利用channelFactory初始化一個NioServerSocketChannel實例,并為其設(shè)置上config中的參數(shù),然后將其注冊到EventLoop中,實際上是委托的channel的Unsafe來實現(xiàn)注冊的,核心邏輯位于 AbstractUnsafe.register0 中 完成注冊
final ChannelFuture initAndRegister() { Channel channel = null; try { //本例子中實際調(diào)用的是 NioServerSocketChannel的構(gòu)造參數(shù), 并為其設(shè)置感興趣的事件類型為 OP_ACCEPT channel = channelFactory.newChannel(); init(channel); } catch (Throwable t) { if (channel != null) { // channel can be null if newChannel crashed (eg SocketException("too many open files")) channel.unsafe().closeForcibly(); } // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t); } ChannelFuture regFuture = config().group().register(channel); if (regFuture.cause() != null) { if (channel.isRegistered()) { channel.close(); } else { channel.unsafe().closeForcibly(); } } return regFuture; }
void init(Channel channel) throws Exception { //設(shè)置屬性 .......... p.addLast(new ChannelInitializer<Channel>() { @Override public void initChannel(final Channel ch) throws Exception { final ChannelPipeline pipeline = ch.pipeline(); ChannelHandler handler = config.handler(); if (handler != null) { pipeline.addLast(handler); } ch.eventLoop().execute(new Runnable() { @Override public void run() { //為NioServerSocketChannel 設(shè)置一個 默認(rèn)的 channelhandler : ServerBootstrapAcceptor , 當(dāng)發(fā)生 accept事件時,將 accept的channel注冊到 childEventLoop中 pipeline.addLast(new ServerBootstrapAcceptor( ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } }); }
private void register0(ChannelPromise promise) { try { // check if the channel is still open as it could be closed in the mean time when the register // call was outside of the eventLoop if (!promise.setUncancellable() || !ensureOpen(promise)) { return; } boolean firstRegistration = neverRegistered; //執(zhí)行channel到 eventloop的 selector doRegister(); neverRegistered = false; registered = true; // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the // user may already fire events through the pipeline in the ChannelFutureListener. pipeline.invokeHandlerAddedIfNeeded(); safeSetSuccess(promise);
//觸發(fā) InboundChannelHnader.channelRegistered 事件 pipeline.fireChannelRegistered();
// Only fire a channelActive if the channel has never been registered. This prevents firing // multiple channel actives if the channel is deregistered and re-registered. if (isActive()) { if (firstRegistration) { //觸發(fā)channelActive事件,并會為 channel 綁定上 read 事件 pipeline.fireChannelActive(); } else if (config().isAutoRead()) { // This channel was registered before and autoRead() is set. This means we need to begin read // again so that we process inbound data. // // See https://github.com/netty/netty/issues/4805 beginRead(); } } } catch (Throwable t) { // Close the channel directly to avoid FD leak. closeForcibly(); closeFuture.setClosed(); safeSetFailure(promise, t); } }
綁定端口邏輯
initAndRegister注冊成功后,開始執(zhí)行真正的綁定端口邏輯,核心邏輯位于 NioSocketChannel.doBind0(SocketAddress) 中
private void doBind0(SocketAddress localAddress) throws Exception { if (PlatformDependent.javaVersion() >= 7) { SocketUtils.bind(javaChannel(), localAddress); } else { SocketUtils.bind(javaChannel().socket(), localAddress); } }
至此 綁定個成功, 當(dāng)觸發(fā) ACCEPT 事件時, 會觸發(fā) NioServerSocketChannel.doReadMessages -> ServerBootstrapAcceptor.channelRead , 并將 子channel 注冊到 childEventLoop中
public void channelRead(ChannelHandlerContext ctx, Object msg) { final Channel child = (Channel) msg; child.pipeline().addLast(childHandler); setChannelOptions(child, childOptions, logger); for (Entry<AttributeKey<?>, Object> e: childAttrs) { child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue()); } try { //注冊channel childGroup.register(child).addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { forceClose(child, future.cause()); } } }); } catch (Throwable t) { forceClose(child, t); } }
以上就是netty服務(wù)端輔助類ServerBootstrap創(chuàng)建邏輯分析的詳細(xì)內(nèi)容,更多關(guān)于netty輔助類ServerBootstrap創(chuàng)建邏輯的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
深入理解SpringBoot中關(guān)于Mybatis使用方法
這篇文章主要介紹了SpringBoot中關(guān)于Mybatis使用方法,非常不錯,具有一定的參考借鑒價值,需要的朋友可以參考下2018-03-03Java并發(fā)編程之volatile與JMM多線程內(nèi)存模型
這篇文章主要介紹了Java并發(fā)volatile與JMM多線程內(nèi)存模型,本文通過實例代碼給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2022-05-05SpringBoot配置 Druid 三種方式(包括純配置文件配置)
本文給大家分享在項目中用純 YML(application.yml 或者 application.properties)文件、Java 代碼配置 Bean 和注解三種方式配置 Alibaba Druid 用于監(jiān)控或者查看 SQL 狀況的相關(guān)知識,感興趣的朋友一起看看吧2021-10-10