Netty服务端启动的整体流程-基于源码4.1.96Final分析

Netty采用的是主从Reactor多线程的模型,参考Scalable IO in Java,但netty的subReactor为一个组

一、从FileServer服务器示例入手 

public final class FileServer {static final boolean SSL = System.getProperty("ssl") != null;// Use the same default port with the telnet example so that we can use the telnet client example to access it.static final int PORT = Integer.parseInt(System.getProperty("port", SSL? "8992" : "8023"));public static void main(String[] args) throws Exception {// Configure SSL.final SslContext sslCtx = ServerUtil.buildSslContext();// Configure the server.主从Reactor线程组EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup)//配置主Reactor中的channel类型.channel(NioServerSocketChannel.class)// 设置主Reactor中channel的option选项,设置底层JDK NIO Socket的一些选项.option(ChannelOption.SO_BACKLOG, 100)//设置主Reactor中Channel->pipline->handler.handler(new LoggingHandler(LogLevel.INFO))//设置 SocketChannel 对应的 Handler;.childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ChannelPipeline p = ch.pipeline();if (sslCtx != null) {p.addLast(sslCtx.newHandler(ch.alloc()));}p.addLast(new StringEncoder(CharsetUtil.UTF_8),new LineBasedFrameDecoder(8192),new StringDecoder(CharsetUtil.UTF_8),new ChunkedWriteHandler(),new FileServerHandler());}});// Start the server.ChannelFuture f = b.bind(PORT).sync();// Wait until the server socket is closed.f.channel().closeFuture().sync();} finally {// Shut down all event loops to terminate all threads.bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

1.1  netty的主从模式    

       首先大致了解netty的主从模式中:bossGroup 中的MainReactor管理的Channel类型为NioServerSocketChannel,用来监听端口,接收客户端连接,为客户端创建初始化NioSocketChannel,然后采用round-robin轮询的方式从workerGroup中选择一个SubReactor与该客户端NioSocketChannel进行绑定。一个SubReactor线程负责处理多个NioSocketChannel上的IO事件

1.2 NioServerSocketChannel

 包含了JDK原生的ServerSocketChannel属性

1.2.1 channel(NioServerSocketChannel.class)

在执行channel的时候,返回的是channelFactory属性,如下:

return channelFactory(new ReflectiveChannelFactory<C>(ObjectUtil.checkNotNull(channelClass, "channelClass")// ReflectiveChannelFactory通过泛型,反射,工厂的方式灵活创建不同类型的channel
public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> {private final Constructor<? extends T> constructor;public ReflectiveChannelFactory(Class<? extends T> clazz) {ObjectUtil.checkNotNull(clazz, "clazz");try {this.constructor = clazz.getConstructor();} catch (NoSuchMethodException e) {throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) +" does not have a public non-arg constructor", e);}}@Overridepublic T newChannel() {try {return constructor.newInstance();} catch (Throwable t) {throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);}}@Overridepublic String toString() {return StringUtil.simpleClassName(ReflectiveChannelFactory.class) +'(' + StringUtil.simpleClassName(constructor.getDeclaringClass()) + ".class)";}
}

1.3 ChannelInitializer的作用

      Pipeline添加ChannelHandler:1、显式添加的方式是由用户在main线程中通过ServerBootstrap#handler的方式添加。2、如果需要添加多个ChannelHandler,则可以通过ChannelInitializer向pipeline中进行添加。

1.3.1 childHandler(new ChannelInitializer<SocketChannel>() {}使用的原因:

      NioSocketChannel是在服务端accept连接后,在服务端NioServerSocketChannel中被创建出来的。但是此时我们正处于配置ServerBootStrap阶段,服务端还没有启动,更没有客户端连接上来,此时客户端NioSocketChannel还没有被创建出来,所以也就没办法向客户端NioSocketChannel的pipeline中添加ChannelHandler。 以及客户端NioSocketChannel中Pipeline里可以添加任意多个ChannelHandler,但是Netty框架无法预知用户到底需要添加多少个ChannelHandler,所以Netty框架提供了回调函数ChannelInitializer#initChannel,使用户可以自定义ChannelHandler的添加行为。

二、服务端启动全过程

    public ChannelFuture bind(int inetPort) {return bind(new InetSocketAddress(inetPort));}public ChannelFuture bind(SocketAddress localAddress) {//校验Netty核心组件是否配置齐全validate();//服务端开始启动,绑定端口地址,接收客户端连接return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));}private ChannelFuture doBind(final SocketAddress localAddress) {//异步创建,初始化,注册ServerSocketChannel到main reactor上final ChannelFuture regFuture = initAndRegister();final Channel channel = regFuture.channel();if (regFuture.cause() != null) {return regFuture;}if (regFuture.isDone()) {   ........serverSocketChannel向Main Reactor注册成功后开始绑定端口....,               } else {//如果此时注册操作没有完成,则向regFuture添加operationComplete回调函数,注册成功后回调。regFuture.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {........serverSocketChannel向Main Reactor注册成功后开始绑定端口...., });return promise;}}

2.1 初始化并注册channel

    final ChannelFuture initAndRegister() {Channel channel = null;try {// io.netty.channel.ReflectiveChannelFactory.newChannelchannel = channelFactory.newChannel();// 初始化channelinit(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 GlobalEventExecutorreturn new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);}// as the Channel is not registered yet we need to force the usage of the GlobalEventExecutorreturn new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);}ChannelFuture regFuture = config().group().register(channel);if (regFuture.cause() != null) {if (channel.isRegistered()) {channel.close();} else {channel.unsafe().closeForcibly();}}// If we are here and the promise is not failed, it's one of the following cases:// 1) If we attempted registration from the event loop, the registration has been completed at this point.//    i.e. It's safe to attempt bind() or connect() now because the channel has been registered.// 2) If we attempted registration from the other thread, the registration request has been successfully//    added to the event loop's task queue for later execution.//    i.e. It's safe to attempt bind() or connect() now://         because bind() or connect() will be executed *after* the scheduled registration task is executed//         because register(), bind(), and connect() are all bound to the same thread.return regFuture;}

2.1.1 channelFactory.newChannel();

    根据1.2.1 可以知道,实际就是调用return constructor.newInstance();也就是实例化NioServerSocketChannel
public class NioServerSocketChannel extends AbstractNioMessageChannelimplements io.netty.channel.socket.ServerSocketChannel {//SelectorProvider(用于创建Selector和Selectable Channels)private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();/*** Create a new instance*/public NioServerSocketChannel() {this(DEFAULT_SELECTOR_PROVIDER);}/*** Create a new instance using the given {@link SelectorProvider}.*/public NioServerSocketChannel(SelectorProvider provider) {this(provider, null);}/*** Create a new instance using the given {@link SelectorProvider} and protocol family (supported only since JDK 15).*/public NioServerSocketChannel(SelectorProvider provider, InternetProtocolFamily family) {this(newChannel(provider, family));}/*** Create a new instance using the given {@link ServerSocketChannel}.*/public NioServerSocketChannel(ServerSocketChannel channel) {super(null, channel, SelectionKey.OP_ACCEPT);config = new NioServerSocketChannelConfig(this, javaChannel().socket());}}

2.1.1.1 SelectorProvider选择器和可选择通道的服务提供者类

    public static SelectorProvider provider() {synchronized (lock) {if (provider != null)return provider;return AccessController.doPrivileged(new PrivilegedAction<SelectorProvider>() {public SelectorProvider run() {if (loadProviderFromProperty())return provider;if (loadProviderAsService())return provider;provider = sun.nio.ch.DefaultSelectorProvider.create();return provider;}});}}// 支持根据系统属性名进行实例化。private static boolean loadProviderFromProperty() {String cn = System.getProperty("java.nio.channels.spi.SelectorProvider");if (cn == null)return false;try {Class<?> c = Class.forName(cn, true,ClassLoader.getSystemClassLoader());provider = (SelectorProvider)c.newInstance();return true;} catch (ClassNotFoundException x) {throw new ServiceConfigurationError(null, x);} catch (IllegalAccessException x) {throw new ServiceConfigurationError(null, x);} catch (InstantiationException x) {throw new ServiceConfigurationError(null, x);} catch (SecurityException x) {throw new ServiceConfigurationError(null, x);}}// 根据spi进行实例化,即META-INF/services/下的定义名为java.nio.channels.spi.SelectorProvider的SPI文件,文件中第一个定义的SelectorProvider实现类全限定名就会被加载。private static boolean loadProviderAsService() {ServiceLoader<SelectorProvider> sl =ServiceLoader.load(SelectorProvider.class,ClassLoader.getSystemClassLoader());Iterator<SelectorProvider> i = sl.iterator();for (;;) {try {if (!i.hasNext())return false;provider = i.next();return true;} catch (ServiceConfigurationError sce) {if (sce.getCause() instanceof SecurityException) {// Ignore the security exception, try the next providercontinue;}throw sce;}}}//因为是windows
public class DefaultSelectorProvider {private DefaultSelectorProvider() {}public static SelectorProvider create() {return new WindowsSelectorProvider();}
}

nio中的channel注册selector 

2.1.1.2 newChannel(provider, family) 

    private static ServerSocketChannel newChannel(SelectorProvider provider, InternetProtocolFamily family) {try {// family为空时 SelectorProviderUtil.newChannel 返回nullServerSocketChannel channel =SelectorProviderUtil.newChannel(OPEN_SERVER_SOCKET_CHANNEL_WITH_FAMILY, provider, family);// 创建 JDK 底层的 ServerSocketChannelreturn channel == null ? provider.openServerSocketChannel() : channel;} catch (IOException e) {throw new ChannelException("Failed to open a socket.", e);}}

因为初始化的时候family为null,所以调用的是JDK底层的openServerSocketChannel

2.1.1.3  NioServerSocketChannel构造

        //设置的是SelectionKey.OP_ACCEPT事件super(null, channel, SelectionKey.OP_ACCEPT);// 创建Channel的配置类NioServerSocketChannelConfig,在配置类中封装了对Channel底层的一些配置行为,以及JDK中的ServerSocket。以及创建NioServerSocketChannel接收数据用的Buffer分配器AdaptiveRecvByteBufAllocatorconfig = new NioServerSocketChannelConfig(this, javaChannel().socket());protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {super(parent);this.ch = ch;this.readInterestOp = readInterestOp;try {// 设置 Channel 为非阻塞模式。ch.configureBlocking(false);} catch (IOException e) {try {ch.close();} catch (IOException e2) {logger.warn("Failed to close a partially initialized socket.", e2);}throw new ChannelException("Failed to enter non-blocking mode.", e);}}protected AbstractChannel(Channel parent) {this.parent = parent;id = newId(); // 全局唯一idunsafe = newUnsafe(); // unsafe 操作底层读写pipeline = newChannelPipeline(); // pipeline 负责业务处理器编排}protected DefaultChannelPipeline newChannelPipeline() {return new DefaultChannelPipeline(this);}protected DefaultChannelPipeline(Channel channel) {this.channel = ObjectUtil.checkNotNull(channel, "channel");succeededFuture = new SucceededChannelFuture(channel, null);voidPromise =  new VoidChannelPromise(channel, true);tail = new TailContext(this);head = new HeadContext(this);head.next = tail;tail.prev = head;}

此时channel的pipeline只有head和tail两个节点;

2.1.2 init(channel);初始化

    @Overridevoid init(Channel channel) {setChannelOptions(channel, newOptionsArray(), logger);setAttributes(channel, newAttributesArray());ChannelPipeline p = channel.pipeline();final EventLoopGroup currentChildGroup = childGroup;final ChannelHandler currentChildHandler = childHandler;final Entry<ChannelOption<?>, Object>[] currentChildOptions = newOptionsArray(childOptions);final Entry<AttributeKey<?>, Object>[] currentChildAttrs = newAttributesArray(childAttrs);//  ChannelInitializer 实现的 initChannel() 方法用于添加 ServerSocketChannel 对应的 Handlerp.addLast(new ChannelInitializer<Channel>() {@Overridepublic void initChannel(final Channel ch) {final ChannelPipeline pipeline = ch.pipeline();// 将handler(new LoggingHandler(LogLevel.INFO)) 中的handler加入pipeLineChannelHandler handler = config.handler();if (handler != null) {pipeline.addLast(handler);}//  然后 Netty 通过异步 task 的方式又向 Pipeline 一个处理器 ServerBootstrapAcceptor,这是一个连接接入器,专门用于接收新的连接,然后把事件分发给 EventLoop 执行ch.eventLoop().execute(new Runnable() {@Overridepublic void run() {pipeline.addLast(new ServerBootstrapAcceptor(ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));}});}});}
  • ServerBootstrapAcceptor:也就是对应MainReactor中的acceptor,本质上也是一种ChannelHandler,主要负责在客户端连接建立好后,初始化客户端NioSocketChannel,在从Reactor线程组中选取一个SubReactor,将客户端NioSocketChannel注册到SubReactor中的selector上。

  • 初始化NioServerSocketChannel中pipeline的时机是:当NioServerSocketChannel注册到Main Reactor之后,绑定端口地址之前,同时为了保证线程安全地初始化pipeline,初始化的动作netty统一交给了Reactor线程进行

  • ServerBootstrapAcceptor 的注册过程为什么又需要封装成异步 task 呢?因为本文案例是

    handler(new LoggingHandler(LogLevel.INFO))但是考虑到过程中可能为new ChannelInitializer<SocketChannel>() ,那么在后续Main Reactor处理register0任务invokeHandlerAddedIfNeeded方法时会调用具体的ChannelInitializer的initChannel方法进行实例会进行添加到最后一个处理节点,如果这里不是异步task那么就会导致该Acceptor为pipeline的一个中间Handler,因此为了保证ServerBootstrapAcceptor是最后一个处理节点,所以本文就封装了一个异步任务。 等到新连接接入时,就可以调用pipeline.fireChannelRead();从head节点依次往下进行传播,直到传播到ServerBootstrapAcceptor

2.2 注册channel到mainReactor中

2.2.1 轮询选取MainReactor

 ChannelFuture regFuture = config().group().register(channel);
从ServerBootstrap获取主Reactor线程组NioEventLoopGroup,将NioServerSocketChannel注册到NioEventLoopGroup中。@Overridepublic ChannelFuture register(Channel channel) {return next().register(channel);}@Overridepublic EventExecutor next() {return chooser.next();}//获取绑定策略@Overridepublic EventExecutorChooser newChooser(EventExecutor[] executors) {// 判断2的次幂if (isPowerOfTwo(executors.length)) {return new PowerOfTwoEventExecutorChooser(executors);} else {return new GenericEventExecutorChooser(executors);}}//采用轮询round-robin的方式选择Reactor@Overridepublic EventExecutor next() {return executors[(int) Math.abs(idx.getAndIncrement() % executors.length)];}private static boolean isPowerOfTwo(int val) {return (val & -val) == val;}正数的补码,反码,原码都是一样的。
负数的补码为反码加1,负数的反码为除符号位原码按位取反。

2.2.2 register

    @Overridepublic ChannelFuture register(Channel channel) {return register(new DefaultChannelPromise(channel, this));}@Overridepublic ChannelFuture register(final ChannelPromise promise) {ObjectUtil.checkNotNull(promise, "promise");promise.channel().unsafe().register(this, promise);return promise;}@Overridepublic final void register(EventLoop eventLoop, final ChannelPromise promise) {ObjectUtil.checkNotNull(eventLoop, "eventLoop");if (isRegistered()) {promise.setFailure(new IllegalStateException("registered to an event loop already"));return;}if (!isCompatible(eventLoop)) {promise.setFailure(new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));return;}//在channel上设置绑定的ReactorAbstractChannel.this.eventLoop = eventLoop;/*** 执行channel注册的操作必须是Reactor线程来完成** 1: 如果当前执行线程是Reactor线程,则直接执行register0进行注册* 2:如果当前执行线程是外部线程,则需要将register0注册操作 封装程异步Task 由Reactor线程执行* */if (eventLoop.inEventLoop()) {register0(promise);} else {//外部线程调用try {eventLoop.execute(new Runnable() {@Overridepublic void run() {register0(promise);}});} catch (Throwable t) {logger.warn("Force-closing a channel whose registration task was not accepted by an event loop: {}",AbstractChannel.this, t);closeForcibly();closeFuture.setClosed();safeSetFailure(promise, t);}}}

当前执行线程并不是Reactor线程,而是用户程序的启动线程Main线程,所以提交异步task并进行了启动Reactor线程

//Reactor线程的启动是在向Reactor提交第一个异步任务的时候启动的。private void execute(Runnable task, boolean immediate) {boolean inEventLoop = inEventLoop();addTask(task);if (!inEventLoop) {startThread();if (isShutdown()) {boolean reject = false;try {if (removeTask(task)) {reject = true;}} catch (UnsupportedOperationException e) {// The task queue does not support removal so the best thing we can do is to just move on and// hope we will be able to pick-up the task before its completely terminated.// In worst case we will log on termination.}if (reject) {reject();}}}if (!addTaskWakesUp && immediate) {wakeup(inEventLoop);}}private void startThread() {if (state == ST_NOT_STARTED) {if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {boolean success = false;try {doStartThread();success = true;} finally {if (!success) {STATE_UPDATER.compareAndSet(this, ST_STARTED, ST_NOT_STARTED);}}}}}

2.2.3 register0()-MainReactor异步任务执行

//一开始Reactor中的任务队列中只有一个任务register0,Reactor线程启动后,会从任务队列中取出任务执行。
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 eventLoopif (!promise.setUncancellable() || !ensureOpen(promise)) {return;}boolean firstRegistration = neverRegistered;// 调用 JDK 底层的 register() 进行注册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中添加的ChannelInitializer的handlerAdded方法,在这里初始化channelPipelinepipeline.invokeHandlerAddedIfNeeded(); // 触发 handlerAdded 事件safeSetSuccess(promise);// channelRegistered 事件是由 fireChannelRegistered() 方法触发,沿着 Pipeline 的 Head 节点传播到 Tail 节点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.//对于服务端ServerSocketChannel来说 只有绑定端口地址成功后 channel的状态才是active的。//此时绑定操作作为异步任务在Reactor的任务队列中,绑定操作还没开始,所以这里的isActive()是false                if (isActive()) {if (firstRegistration) {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/4805beginRead();}}} catch (Throwable t) {// Close the channel directly to avoid FD leak.closeForcibly();closeFuture.setClosed();safeSetFailure(promise, t);}}

2.2.3.1 doRegister

    @Overrideprotected void doRegister() throws Exception {boolean selected = false;for (;;) {try {selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);return;} catch (CancelledKeyException e) {if (!selected) {// Force the Selector to select now as the "canceled" SelectionKey may still be// cached and not removed because no Select.select(..) operation was called yet.eventLoop().selectNow();selected = true;} else {// We forced a select operation on the selector before but the SelectionKey is still cached// for whatever reason. JDK bug ?throw e;}}}}public final SelectionKey register(Selector sel, int ops,Object att)throws ClosedChannelException{synchronized (regLock) {if (!isOpen())throw new ClosedChannelException();if ((ops & ~validOps()) != 0)throw new IllegalArgumentException();if (blocking)throw new IllegalBlockingModeException();SelectionKey k = findKey(sel);if (k != null) {k.interestOps(ops);k.attach(att);}if (k == null) {// New registrationsynchronized (keyLock) {if (!isOpen())throw new ClosedChannelException();k = ((AbstractSelector)sel).register(this, ops, att);addKey(k);}}return k;}}

       javaChannel().register() 负责调用 JDK 底层,将 Channel 注册到 Selector 上,register() 的第三个入参传入的是 Netty 自己实现的 Channel 对象,调用 register() 方法会将它绑定在 JDK 底层 Channel 的attachment上。这样在每次 Selector 对象进行事件循环时,Netty 都可以从返回的 JDK 底层 Channel 中获得自己的 Channel 对象。

2.2.3.2 handlerAdded

初始化ChannelPipeline的时机是当Channel向对应的Reactor注册成功后,在handlerAdded事件回调中利用ChannelInitializer进行初始化。

io.netty.channel.ChannelInitializer#handlerAdded

    @Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {// 必须是注册以后if (ctx.channel().isRegistered()) {// This should always be true with our current DefaultChannelPipeline implementation.// The good thing about calling initChannel(...) in handlerAdded(...) is that there will be no ordering// surprises if a ChannelInitializer will add another ChannelInitializer. This is as all handlers// will be added in the expected order.if (initChannel(ctx)) {// We are done with init the Channel, removing the initializer now.removeState(ctx);}}}//ChannelInitializer实例是被所有的Channel共享的,用于初始化ChannelPipeline//通过Set集合保存已经初始化的ChannelPipeline,避免重复初始化同一ChannelPipelineprivate final Set<ChannelHandlerContext> initMap = Collections.newSetFromMap(new ConcurrentHashMap<ChannelHandlerContext, Boolean>());private boolean initChannel(ChannelHandlerContext ctx) throws Exception {if (initMap.add(ctx)) { // Guard against re-entrance.try {initChannel((C) ctx.channel());} catch (Throwable cause) {exceptionCaught(ctx, cause);} finally {ChannelPipeline pipeline = ctx.pipeline();if (pipeline.context(this) != null) {//初始化完毕后,从pipeline中移除自身pipeline.remove(this);}}return true;}return false;}//匿名类实现,这里指定具体的初始化逻辑protected abstract void initChannel(C ch) throws Exception;private void removeState(final ChannelHandlerContext ctx) {//从initMap防重Set集合中删除ChannelInitializerif (ctx.isRemoved()) {initMap.remove(ctx);} else {ctx.executor().execute(new Runnable() {@Overridepublic void run() {initMap.remove(ctx);}});}}

执行完成后pipeline如下:

执行完整个 register0() 的注册流程之后,EventLoop 线程会将 ServerBootstrapAcceptor 添加到 Pipeline 当中(提交的任务执行)

ch.eventLoop().execute(new Runnable() {@Overridepublic void run() {pipeline.addLast(new ServerBootstrapAcceptor(ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));}}

也就是统一在EventLoop 线程中初始化pipeLine,保证线程安全

2.2.4 Reactor线程触发注册成功safeSetSuccess(promise);

Reactor设置注册成功后,启动线程监听到完成任务,那么就进行接下来的绑定端口操作

        // 若执行完毕进行端口绑定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() {@Overridepublic 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/2586promise.registered();doBind0(regFuture, channel, localAddress, promise);}}});return promise;}

2.3 端口绑定

    private static void doBind0(final ChannelFuture regFuture, final Channel channel,final SocketAddress localAddress, final ChannelPromise promise) {// This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up// the pipeline in its channelRegistered() implementation.channel.eventLoop().execute(new Runnable() {@Override
sss if (regFuture.isSuccess()) {channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);} else {promise.setFailure(regFuture.cause());}}});}

提交到异步任务到Reactor,绑定逻辑需要注册逻辑处理完之后运行,如上面的ServerBootstrapAcceptor异步任务执行完

    @Overridepublic ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {return pipeline.bind(localAddress, promise);}@Overridepublic final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {return tail.bind(localAddress, promise);}@Overridepublic ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {ObjectUtil.checkNotNull(localAddress, "localAddress");if (isNotValidPromise(promise, false)) {// cancelledreturn promise;}final AbstractChannelHandlerContext next = findContextOutbound(MASK_BIND);EventExecutor executor = next.executor();if (executor.inEventLoop()) {next.invokeBind(localAddress, promise);} else {safeExecute(executor, new Runnable() {@Overridepublic void run() {next.invokeBind(localAddress, promise);}}, promise, null, false);}return promise;}private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {if (invokeHandler()) {try {// DON'T CHANGE// Duplex handlers implements both out/in interfaces causing a scalability issue// see https://bugs.openjdk.org/browse/JDK-8180450final ChannelHandler handler = handler();final DefaultChannelPipeline.HeadContext headContext = pipeline.head;if (handler == headContext) {headContext.bind(this, localAddress, promise);} else if (handler instanceof ChannelDuplexHandler) {((ChannelDuplexHandler) handler).bind(this, localAddress, promise);} else {((ChannelOutboundHandler) handler).bind(this, localAddress, promise);}} catch (Throwable t) {notifyOutboundHandlerException(t, promise);}} else {bind(localAddress, promise);}}

调用pipeline.bind(localAddress, promise)pipeline中传播bind事件,触发回调pipeline中所有ChannelHandlerbind方法

事件在pipeline中的传播具有方向性:

  • inbound事件HeadContext开始逐个向后传播直到TailContext

  • outbound事件则是反向传播,从TailContext开始反向向前传播直到HeadContext

  •     private AbstractChannelHandlerContext findContextOutbound(int mask) {AbstractChannelHandlerContext ctx = this;EventExecutor currentExecutor = executor();do {ctx = ctx.prev;
    // 跳过了 private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter} while (skipContext(ctx, currentExecutor, mask, MASK_ONLY_OUTBOUND));return ctx;}private static boolean skipContext(AbstractChannelHandlerContext ctx, EventExecutor currentExecutor, int mask, int onlyMask) {// Ensure we correctly handle MASK_EXCEPTION_CAUGHT which is not included in the MASK_EXCEPTION_CAUGHTreturn (ctx.executionMask & (onlyMask | mask)) == 0 ||// We can only skip if the EventExecutor is the same as otherwise we need to ensure we offload// everything to preserve ordering.//// See https://github.com/netty/netty/issues/10067(ctx.executor() == currentExecutor && (ctx.executionMask & mask) == 0);}

通过上面代码可以知道bind事件在Netty中被定义为outbound事件,所以它在pipeline中是反向传播。先从TailContext开始反向传播直到HeadContext

因此bind的核心逻辑也正是实现在HeadContext中。

headContext.bind(this, localAddress, promise);

底层实际就是

        @Overridepublic void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {unsafe.bind(localAddress, promise);}@Overridepublic final void bind(final SocketAddress localAddress, final ChannelPromise promise) {assertEventLoop();if (!promise.setUncancellable() || !ensureOpen(promise)) {return;}// See: https://github.com/netty/netty/issues/576if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&localAddress instanceof InetSocketAddress &&!((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&!PlatformDependent.isWindows() && !PlatformDependent.maybeSuperUser()) {// Warn a user about the fact that a non-root user can't receive a// broadcast packet on *nix if the socket is bound on non-wildcard address.logger.warn("A non-root user can't receive a broadcast packet if the socket " +"is not bound to a wildcard address; binding to a non-wildcard " +"address (" + localAddress + ") anyway as requested.");}boolean wasActive = isActive();try {doBind(localAddress);} catch (Throwable t) {safeSetFailure(promise, t);closeIfClosed();return;}if (!wasActive && isActive()) {invokeLater(new Runnable() {@Overridepublic void run() {pipeline.fireChannelActive();}});}safeSetSuccess(promise);}@SuppressJava6Requirement(reason = "Usage guarded by java version check")@Overrideprotected void doBind(SocketAddress localAddress) throws Exception {if (PlatformDependent.javaVersion() >= 7) {javaChannel().bind(localAddress, config.getBacklog());} else {javaChannel().socket().bind(localAddress, config.getBacklog());}}

Netty 会根据 JDK 版本的不同,分别调用 JDK 底层不同的 bind() 方法。我使用的是 JDK8,所以会调用 JDK 原生 Channel 的 bind() 方法。执行完 doBind() 之后,服务端 JDK 原生的 Channel 真正已经完成端口绑定了。

2.3.1 判断是否为激活

    @Overridepublic boolean isActive() {// As java.nio.ServerSocketChannel.isBound() will continue to return true even after the channel was closed// we will also need to check if it is open.return isOpen() && javaChannel().socket().isBound();}

2.3.2 channelActive事件

   完成端口绑定之后,Channel 处于活跃 Active 状态,然后会调用 pipeline.fireChannelActive() 方法触发 channelActive 事件。

    channelActive事件在Netty中定义为inbound事件,所以它在pipeline中的传播为正向传播,从HeadContext一直到TailContext为止。

channelActive事件回调中需要触发向Selector指定需要监听的IO事件~~OP_ACCEPT事件

    @Overridepublic final ChannelPipeline fireChannelActive() {AbstractChannelHandlerContext.invokeChannelActive(head);return this;}private void invokeChannelActive() {if (invokeHandler()) {try {// DON'T CHANGE// Duplex handlers implements both out/in interfaces causing a scalability issue// see https://bugs.openjdk.org/browse/JDK-8180450final ChannelHandler handler = handler();final DefaultChannelPipeline.HeadContext headContext = pipeline.head;if (handler == headContext) {headContext.channelActive(this);} else if (handler instanceof ChannelDuplexHandler) {((ChannelDuplexHandler) handler).channelActive(this);} else {((ChannelInboundHandler) handler).channelActive(this);}} catch (Throwable t) {invokeExceptionCaught(t);}} else {fireChannelActive();}}@Overridepublic void channelActive(ChannelHandlerContext ctx) {//pipeline中继续向后传播channelActive事件ctx.fireChannelActive();//如果是autoRead 则自动触发read事件传播//在read回调函数中 触发OP_ACCEPT注册readIfIsAutoRead();}private void readIfIsAutoRead() {if (channel.config().isAutoRead()) {channel.read();}}
    @Overridepublic Channel read() {pipeline.read();return this;}@Overridepublic final ChannelPipeline read() {tail.read();return this;}@Overridepublic void read(ChannelHandlerContext ctx) {unsafe.beginRead();}@Overridepublic final void beginRead() {assertEventLoop();try {doBeginRead();} catch (final Exception e) {invokeLater(new Runnable() {@Overridepublic void run() {pipeline.fireExceptionCaught(e);}});close(voidPromise());}}@Overrideprotected void doBeginRead() throws Exception {// Channel.read() or ChannelHandlerContext.read() was calledfinal SelectionKey selectionKey = this.selectionKey;if (!selectionKey.isValid()) {return;}readPending = true;final int interestOps = selectionKey.interestOps();if ((interestOps & readInterestOp) == 0) {selectionKey.interestOps(interestOps | readInterestOp);}}

在执行完 channelActive 事件传播之后,会调用 readIfIsAutoRead() 方法触发 Channel 的 read 事件,而它最终调用到 AbstractNioChannel 中的 doBeginRead() 方法,其中 readInterestOp 参数就是在前面初始化 Channel 所传入的 SelectionKey.OP_ACCEPT 事件,所以 OP_ACCEPT 事件会被注册到 Channel 的事件集合中。

2.4 服务启动总结

  • 创建服务端 Channel:本质是创建 JDK 底层原生的 Channel,并初始化几个重要的属性,包括 id、unsafe、pipeline 等。
  • 初始化服务端 Channel:设置 Socket 参数以及用户自定义属性,并添加两个特殊的处理器 ChannelInitializer 和 ServerBootstrapAcceptor。
  • 注册服务端 Channel:调用 JDK 底层将 Channel 注册到 Selector 上。
  • 端口绑定:调用 JDK 底层进行端口绑定,并触发 channelActive 事件,把 OP_ACCEPT 事件注册到 Channel 的事件集合中。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/69694.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【笔记】软件测试的艺术

软件测试的心理学和经济学 测试是为发现错误而执行程序的过程&#xff0c;所以它是一个破坏性的过程&#xff0c;测试是一个“施虐”的过程。 软件测试的10大原则 1、测试用例需要对预期输出的结果有明确的定义 做这件事的前提是能够提前知晓需求和效果图&#xff0c;如果不…

Linux 操作系统实战视频课 - GPIO 基础介绍

文章目录 一、GPIO 概念说明二、视频讲解沉淀、分享、成长,让自己和他人都能有所收获!😄 📢本篇我们将讲解 GPIO 。 一、GPIO 概念说明 ARM 平台中的 GPIO(通用输入/输出)是用于与外部设备进行数字输入和输出通信的重要硬件接口。ARM 平台的 GPIO 特性可以根据具体的芯…

C++11新特性① | C++11 常用关键字实战详解

目录 1、引言 2、C11 新增关键字详解 2.1、auto 2.2、override 2.3、final 2.4、nullptr 2.5、使用delete阻止拷贝类对象 2.6、decltype 2.7、noexcept 2.8、constexpr 2.9、static_assert VC常用功能开发汇总&#xff08;专栏文章列表&#xff0c;欢迎订阅&#xf…

LeetCode 剑指 Offer 10- I. 斐波那契数列

LeetCode 剑指 Offer 10- I. 斐波那契数列 题目描述 写一个函数&#xff0c;输入 n &#xff0c;求斐波那契&#xff08;Fibonacci&#xff09;数列的第 n 项&#xff08;即 F(N)&#xff09;。斐波那契数列的定义如下&#xff1a; F(0) 0, F(1) 1 F(N) F(N - 1) F(N - …

Unity项目包体优化经验方法论(Android平台)

前言 本篇文章主要讲解对于Unity Android平台也就是APK包体的优化经验&#xff0c;使用哪些工具能够更加便利的定位资源重灾区。本篇讲解的方法中对于Unity资源使用的AssetBundle的方式&#xff0c;如果使用addressable或其他资源管理方式&#xff0c;我还不是很清楚是否适用&…

使用pyenv安装python缓慢或无法安装

使用pyenv安装python缓慢或无法安装 这一定程度上和网络情况有关&#xff0c;下面提供几个常见方法&#xff1a; 关闭 VPN 后重新安装使用管理员权限打开命令窗口后安装如下 手动安装 pyenv 在执行 pyenv install --- 命令的时候&#xff0c;会连接远程库&#xff0c;将要安…

matlab使用教程(28)—微分方程(ODE)求解常见问题

1.非负 ODE 解 本博客说明如何将 ODE 解约束为非负解。施加非负约束不一定总是可有可无&#xff0c;在某些情况下&#xff0c;由于方程的物理解释或解性质的原因&#xff0c;可能有必要施加非负约束。仅在必要时对解施加此约束&#xff0c;例如不这样做积分就会失败或者解将不…

springboot项目中application.properties无法变成小树叶问题解决

1.检查我们的resources目录的状态&#xff0c;看看是不是处在普通文件夹的状态&#xff0c;如果是的话&#xff0c;我们需要重新mark一下 右键点击文件夹&#xff0c;选择mark directory as → resources root 此时我们发现配置文件变成了小树叶 2.如果执行了上述方法还是不行…

智能手机收入和出货量双双下滑,造车成本不断增长,小米集团仍面临风险

来源&#xff1a;猛兽财经 作者&#xff1a;猛兽财经 华尔街分析师对小米集团第二季度的业绩预测 在8月29日小米集团&#xff08;01810&#xff09;公布其2023年第二季度财报之前&#xff0c;华尔街分析师曾预测该公司第二季度的业绩将超出2023年第一季度的业绩。 根据S&P …

uniapp点击事件在小程序中无法传参

这个问题很是神奇&#xff0c;第一次遇到。在h5中&#xff0c;点击事件可以正常传参&#xff0c;打包小程序后确失效了。 修改&#xff1a;for循环中的key&#xff0c;使用 index就好了

计算机竞赛 基于深度学习的人脸表情识别

文章目录 0 前言1 技术介绍1.1 技术概括1.2 目前表情识别实现技术 2 实现效果3 深度学习表情识别实现过程3.1 网络架构3.2 数据3.3 实现流程3.4 部分实现代码 4 最后 0 前言 &#x1f525; 优质竞赛项目系列&#xff0c;今天要分享的是 基于深度学习的人脸表情识别 该项目较…

Web前端自动化测试Cypress实践总结

本文主要首先主要介绍了什么是自动化测试&#xff0c;接着对常用的自动化测试框架进行了对比分析&#xff0c;最后&#xff0c;介绍了如果将自动化测试框架Cypress运用在项目中。 一、自动化测试概述 为了保障软件质量&#xff0c;并减少重复性的测试工作&#xff0c;自动化测…

Java中网络的基本介绍。网络通信,网络,ip地址,域名,端口,网络通信协议,TCP/IP传输过程,网络通信协议模型,TCP协议,UDP协议

- 网络通信 概念&#xff1a;网络通信是指通过计算机网络进行信息传输的过程&#xff0c;包括数据传输、语音通话、视频会议等。在网络通信中&#xff0c;数据被分成一系列的数据包&#xff0c;并通过网络传输到目的地。在数据传输过程中&#xff0c;需要确保数据的完整性、准…

1773_把vim的tab键设置为4个空格显示

全部学习汇总&#xff1a; GitHub - GreyZhang/editors_skills: Summary for some common editor skills I used. 有时候自己觉得自己很奇怪&#xff0c;看着Linux的命令窗口就觉得很顺眼。那些花花绿绿的字符以及繁多的方便命令工具&#xff0c;确实是比Windows强不少。不过&a…

FBX SDK 开发环境配置 visual studio 2022

FBX | Adaptable File Formats for 3D Animation Software | Autodesk. 下载windows的sdk并安装. 创建一个c console 工程 设置include目录 添加预处理宏 FBX_SHARED1 添加fbx sdk lib 目录 添加依赖lib : libfbxsdk-md.lib libxml2-md.lib zlib-md.lib 配置完毕.

JS 方法实现复制粘贴

背景 以前我们一涉及到复制粘贴功能&#xff0c;实现思路一般都是&#xff1a; 创建一个 textarea 标签 让这个 textarea 不可见&#xff08;定位&#xff09; 给这个 textarea 赋值 把这个 textarea 塞到页面中 调用 textarea 的 select 方法 调用 document.execCommand…

go锁-waitgroup

如果被等待的协程没了&#xff0c;直接返回 否则&#xff0c;waiter加一&#xff0c;陷入sema add counter 被等待协程没做完&#xff0c;或者没人在等待&#xff0c;返回 被等待协程都做完&#xff0c;且有人在等待&#xff0c;唤醒所有sema中的协程 WaitGroup实现了一组协程…

TCP的滑动窗口协议有什么用?

分析&回答 滑动窗口协议&#xff1a; TCP协议的使用维持发送方/接收方缓冲区 缓冲区是 用来解决网络之间数据不可靠的问题&#xff0c;例如丢包&#xff0c;重复包&#xff0c;出错&#xff0c;乱序 在TCP协议中&#xff0c;发送方和接受方通过各自维护自己的缓冲区。通…

csharp开发日常之Activator.CreateInstance构造函数生成实例

目录 一、需求&#xff1a;项目中经常需要动态生成对象&#xff0c;而非采用new关键字方式&#xff0c;例如Java里面的根据类全限定名反射生成对象实例。 二、方案&#xff1a;采用Activator.CreateInstance 三、代码例子演示 1、代码结构 2、创建接口 3、创建IObjcet接口的…

FPN模型

【简介】 2017年&#xff0c;T.-Y.Lin等人在Faster RCNN的基础上进一步提出了特征金字塔网络FPN(Feature Pyramid Networks)技术。在FPN技术出现之前&#xff0c;大多数检测算法的检测头都位于网络的最顶层(最深层)&#xff0c;虽说最深层的特征具备更丰富的语义信息&#xff0…