直播网站怎么做的,网站域名免费吗,网站企业地图,自己免费建设网站highlight: arduino-light 服务端启动主要流程 •创建 selector •创建 server socket channel •初始化 server socket channel •给 server socket channel 从 boss group 中选择一个 NioEventLoop •将 server socket channel 注册到选择的 NioEventLoop 的 selector •… highlight: arduino-light 服务端启动主要流程 •创建 selector •创建 server socket channel •初始化 server socket channel •给 server socket channel 从 boss group 中选择一个 NioEventLoop •将 server socket channel 注册到选择的 NioEventLoop 的 selector •绑定地址启动 •注册接受连接事件(OP_ACCEPT)到 selector上 从Echo服务器示例入手 在《引导器作用客户端和服务端启动都要做些什么》的课程中我们介绍了如何使用引导器搭建服务端的基本框架。在这里我们实现了一个最简单的 Echo 服务器用于调试 Netty 服务端启动的源码。 java public class EchoServer { public void startEchoServer(int port) throws Exception { //默认会创建 cpu核心数*2 个 NioEventLoop //创建NioEventLoop 会调用openSelector 创建一个Selector //将创建的Selector赋值给NioEventLoop的selector属性 EventLoopGroup bossGroup new NioEventLoopGroup(); EventLoopGroup workerGroup new NioEventLoopGroup(); try { ServerBootstrap b new ServerBootstrap(); b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializerSocketChannel() { Override public void initChannel(SocketChannel ch) { ch.pipeline().addLast (new FixedLengthFrameDecoder(10)); ch.pipeline().addLast(new ResponseSampleEncoder()); ch.pipeline().addLast(new RequestSampleHandler()); } }); //同步阻塞 ChannelFuture f b.bind(port).sync(); f.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { new EchoServer().startEchoServer(8088); } } public class ResponseSampleEncoder extends MessageToByteEncoderResponseSample { Override protected void encode(ChannelHandlerContext ctx, ResponseSample msg, ByteBuf out) { if (msg ! null) { out.writeBytes(msg.getCode().getBytes()); out.writeBytes(msg.getData().getBytes()); out.writeLong(msg.getTimestamp()); } } } public class RequestSampleHandler extends ChannelInboundHandlerAdapter { Override public void channelRead(ChannelHandlerContext ctx, Object msg) { String data ((ByteBuf) msg).toString(CharsetUtil.UTF_8); ResponseSample response new ResponseSample(OK, data, System.currentTimeMillis()); ctx.channel().writeAndFlush(response); } } java 我们以引导器 ServerBootstrap 为切入点开始深入分析 Netty 服务端的启动流程。在服务端启动之前需要配置 ServerBootstrap 的相关参数这一步大致可以分为以下几个步骤 配置 EventLoopGroup 线程组 配置 Channel 的类型 设置初始化Handler 设置网络监听的端口 设置处理Handler 配置 Channel 参数。 配置 ServerBootstrap 参数的过程非常简单把参数值保存在 ServerBootstrap 定义的成员变量里就可以了。我们可以看下 ServerBootstrap 的成员变量定义基本与 ServerBootstrap 暴露出来的配置方法是一一对应的。如下所示我以注释的形式说明每个成员变量对应的调用方法。 java volatile EventLoopGroup group; // group() volatile EventLoopGroup childGroup; // group() volatile ChannelFactory? extends C channelFactory; // channel() volatile SocketAddress localAddress; // localAddress MapChannelOption?, Object childOptions new ConcurrentHashMapChannelOption?, Object(); // childOption() volatile ChannelHandler childHandler; // childHandler() ServerBootstrapConfig config new ServerBootstrapConfig(this); 关于 ServerBootstrap 如何为每个成员变量保存参数的过程我们就不一一展开了你可以理解为这部分工作只是一个前置准备课后你可以自己跟进下每个方法的源码。 今天我们核心聚焦在 b.bind().sync() 这行代码bind() 才是真正进行服务器端口绑定和启动的入口sync() 表示阻塞等待服务器启动完成。接下来我们对 bind() 方法进行展开分析。 在开始源码分析之前我们带着以下几个问题边看边思考 Netty 自己实现的 Channel 与 JDK 底层的 Channel 是如何产生联系的 ChannelInitializer 这个特殊的 Handler 处理器的作用是什么 Pipeline 初始化的过程是什么样的 服务端启动全过程 入口:bind方法 首先我们来看下 ServerBootstrap 中 bind() 方法的源码实现 java //bind方法入口 public ChannelFuture bind() { validate(); //已通过 SocketAddress localAddress this.localAddress; if (localAddress null) { throw new IllegalStateException(localAddress not set); } //doBind return doBind(localAddress); } java private ChannelFuture doBind(final SocketAddress localAddress) { //初始化并注册 Channel同时返回一个 ChannelFuture 实例 regFuture //所以我们可以猜测出 //initAndRegister() 是一个异步的过程。 final ChannelFuture regFuture initAndRegister(); final Channel channel regFuture.channel();
//判断 initAndRegister() 的过程是否发生异常如果发生异常则直接返回。
if (regFuture.cause() ! null) {return regFuture;
} //regFuture.isDone() 即initAndRegister() 是否执行完毕 //如果执行完毕则调用doBind0() 进行Socket 绑定 //如果 initAndRegister() 还没有执行结束regFuture 会添加ChannelFutureListener 回调监听 //当 initAndRegister() 执行结束后会调用 operationComplete()同样通过 doBind0() 进行端口绑定。 if (regFuture.isDone()) { ChannelPromise promise channel.newPromise(); //doBind0() doBind0(regFuture, channel, localAddress, promise); return promise; } else { 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) { promise.setFailure(cause); } else { promise.registered(); //regFuture.isDone() 如果没有执行完毕,这里的监听器会调用doBind0 doBind0(regFuture, channel, localAddress, promise); } } }); return promise; } } doBind0() 整个实现结构非常清晰其中 initAndRegister() 负责 NioSocketServerChannel初始化和注册doBind0() 用于端口绑定。这两个过程最为重要下面我们分别进行详细的介绍。 InitAndRegister方法 initAndRegister() 方法顾名思义主要负责创建NioServerSocketChannel初始化NioServerSocketChannel和注册NioServerSocketChannel的相关工作我们具体看下它的源码实现 java final ChannelFuture initAndRegister() { Channel channel null; try { // 通过指定的NioSocketServerChannel.class 反射创建 // 1.创建 Channel channel channelFactory.newChannel(); // 2.初始化 Channel init(channel); } catch (Throwable t) { if (channel ! null) { channel.unsafe().closeForcibly(); return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t); } return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t); } //3.注册Channel //config().group()是什么 //是bossGroup ChannelFuture regFuture config().group().register(channel); if (regFuture.cause() ! null) { if (channel.isRegistered()) { channel.close(); } else { channel.unsafe().closeForcibly(); } } return regFuture; } initAndRegister() 可以分为三步创建 Channel、初始化 Channel 和注册 Channel接下来我们一步步进行拆解分析。 创建服务器端Channel 1.创建NioServerSocketChannel 首先看下创建 Channel 的过程直接跟进 channelFactory.newChannel() 的源码。 java public class ReflectiveChannelFactoryT extends Channel implements ChannelFactoryT { private final Constructor? extends T constructor; public ReflectiveChannelFactory(Class? extends T clazz) { ObjectUtil.checkNotNull(clazz, clazz); try { //这里通过泛型反射工厂 获取无参构造方法 //传进来的clazz是NioServerSocketChannel.class this.constructor clazz.getConstructor(); } catch (NoSuchMethodException e) { throw new IllegalArgumentException(Class StringUtil.simpleClassName(clazz) does not have a public non-arg constructor, e); } } Override public T newChannel() { try { // 反射创建对象 return constructor.newInstance(); } catch (Throwable t) { throw new ChannelException(Unable to create Channel from class constructor.getDeclaringClass(), t); } } // 省略其他代码 } 在前面 EchoServer的示例中我们通过 channel(NioServerSocketChannel.class) 配置 Channel 的类型工厂类 ReflectiveChannelFactory 是在该过程中被创建的。 从 constructor.newInstance() 我们可以看出ReflectiveChannelFactory 通过反射创建出 NioServerSocketChannel 对象所以我们重点需要关注 NioServerSocketChannel 的构造函数。 java //private static final SelectorProvider //DEFAULTSELECTORPROVIDER SelectorProvider.provider(); public NioServerSocketChannel() { //DEFAULTSELECTORPROVIDER: 根据不同的系统返回不同的SelectorProvider // 很熟悉啊newSocket(DEFAULTSELECTORPROVIDER)是创建 JDK 底层的 ServerSocketChannel this(newSocket(DEFAULTSELECTORPROVIDER)); } //根据不同的 SelectorProvider 创建不同的JDK 底层的 ServerSocketChannel private static ServerSocketChannel newSocket(SelectorProvider provider) { try { // 创建 JDK 底层的 ServerSocketChannel 实现类是ServerSocketChannelImpl return provider.openServerSocketChannel(); } catch (IOException e) { throw new ChannelException( Failed to open a server socket., e); } } //将JDK 底层的 ServerSocketChannel 包装为NioServerSocketChannel 并注册OPACCEPT事件 public NioServerSocketChannel(ServerSocketChannel channel) { // 调用父类AbstractChannel方法 // 注意这里是SelectionKey.OPACCEPT16 // 并不是注册ACCEPT事件 // 只是设置 this.readInterestOp readInterestOp; 即设置为16 super(null, channel, SelectionKey.OP_ACCEPT); //创建channel时已经创建了config config new NioServerSocketChannelConfig(this, javaChannel().socket()); } SelectorProvider 是 JDK NIO 中的抽象类实现通过 openServerSocketChannel() 方法可以用于创建服务端的 ServerSocketChannel。而且 SelectorProvider 会根据操作系统类型和版本的不同返回不同的实现类具体可以参考 DefaultSelectorProvider 的源码实现 java public static SelectorProvider provider() { synchronized (lock) { if (provider ! null) return provider; return AccessController.doPrivileged( new PrivilegedActionSelectorProvider() { public SelectorProvider run() { //1.读取配置根据配置的class获取provider 獲取不到到第二步 if (loadProviderFromProperty()) return provider; //2.通过spi获取provider 获取不到到第三步 if (loadProviderAsService()) return provider; //3.DefaultSelectorProvider#create创建provider //根据不同的系统创建不同的Selector 或者是说jdk不同 //Linux下JDK的下载和安装与Windows下并没有太大的不同 //只是对一些环境的设置稍有不同。 //在windows环境下的是WindowsSelectorProvider provider sun.nio.ch.DefaultSelectorProvider.create(); return provider; } }); } } 看一下sun.nio.ch.DefaultSelectorProvider.create(); java public static SelectorProvider create() { String osname AccessController .doPrivileged(new GetPropertyAction(os.name)); if (osname.equals(SunOS)) return createProvider(sun.nio.ch.DevPollSelectorProvider); if (osname.equals(Linux)) return createProvider(sun.nio.ch.EPollSelectorProvider); //默认返回的是PollSelectorProvider return new sun.nio.ch.PollSelectorProvider(); } 在这里我们只讨论 Linux 操作系统的场景在 Linux 内核 2.6版本及以上都会默认采用 EPollSelectorProvider。 如果是旧版本则使用 PollSelectorProvider。对于目前的主流 Linux 平台而言都是采用 Epoll 机制实现的。 创建完JDK的ServerSocketChannel我们回到 NioServerSocketChannel 的构造函数 java //将JDK 底层的 ServerSocketChannel 包装为NioServerSocketChannel 并注册OP_ACCEPT事件 public NioServerSocketChannel(ServerSocketChannel channel) { // 调用父类AbstractChannel方法 // 注意这里是SelectionKey.OP_ACCEPT16 // 并不是注册ACCEPT事件 // 只是设置 this.readInterestOp readInterestOp; 即设置为16 super(null, channel, SelectionKey.OP_ACCEPT); //创建channel时已经创建了config config new NioServerSocketChannelConfig(this, javaChannel().socket()); } 接着它会通过依次调用到父类的构造进行初始化工作 java super(null, channel, SelectionKey.OP_ACCEPT); 最终我们可以定位到 AbstractNioChannel 和 AbstractChannel 的构造函数 java protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) { super(parent); // 省略其他代码 //设置为16 this.readInterestOp readInterestOp; try { //设置非阻塞 ch.configureBlocking(false); } catch (IOException e) { // 省略其他代码 } } protected AbstractChannel(Channel parent) { this.parent parent; // Channel 全局唯一 id id newId(); // unsafe 操作底层读写 unsafe newUnsafe(); // pipeline 负责业务处理器编排 // 会初始化TailContext和HeadContext pipeline newChannelPipeline(); } 2.设置pipeline 首先调用 AbstractChannel 的构造函数创建了三个重要的成员变量分别为 id、unsafe、pipeline。 id 表示全局唯一的 Channel unsafe 用于操作底层数据的读写操作 pipeline 负责业务处理器的编排。 3.设置非阻塞模式 初始化状态pipeline 的内部结构只包含头尾两个节点如下图所示。三个核心成员变量创建好之后会回到 AbstractNioChannel 的构造函数通过 ch.configureBlocking(false) 设置 Channel 是非阻塞模式。 创建服务端 Channel 的过程我们已经讲完了简单总结下其中几个重要的步骤 java ReflectiveChannelFactory 通过反射创建 NioServerSocketChannel 实例 创建JDK底层的ServerSocketChannel,调用NioServerSocketChannel的构造函数,将底层的ServerSocketChannel包装为 NioServerSocketChannel。 在构造函数中,执行以下逻辑 1.为 Channel 创建 id、unsafe、pipeline 三个重要的成员变量 2.设置 Channel 为非阻塞模式。 返回 NioServerSocketChannel。 初始化服务器端Channel 回到 ServerBootstrap 的 initAndRegister() 方法继续跟进用于初始化服务端 Channel 的 init() 方法源码 注意Channel是NioServerSocketChannel java void init(Channel channel) { // 设置 Socket 参数,用户自定义的参数都放在了名为options的map中,此处是遍历map设置属性 // 底层也是做判断设置的 好low啊 setChannelOptions (channel, options0().entrySet().toArray(newOptionArray(0)), logger); // 保存用户自定义属性 setAttributes(channel, attrs0().entrySet().toArray(newAttrArray(0))); //获取pipeline ChannelPipeline p channel.pipeline(); //下面的四个参数是为了childOption 即 NioSocketChannel // 也是获取 ServerBootstrapAcceptor 的构造参数 final EventLoopGroup currentChildGroup childGroup; final ChannelHandler currentChildHandler childHandler; final EntryChannelOption?, Object[] currentChildOptions childOptions.entrySet().toArray(newOptionArray(0)); final EntryAttributeKey?, Object[] currentChildAttrs childAttrs.entrySet().toArray(newAttrArray(0)); //class ChannelInitializer extends ChannelInboundHandlerAdapter //ChannelInitializer是一个ChannelInboundHandlerAdapter //添加特殊的Handler处理器ChannelInitializer 它是一个一次性初始化的hanlder //负责添加一个ServerBootstrapAcceptor handler添加完后自己就移除了: //ServerBootstrapAcceptor handler 负责接收客户端连接创建连接后对连接的初始化工作。 p.addLast(new ChannelInitializerChannel() { Override public void initChannel(final Channel ch) { final ChannelPipeline pipeline ch.pipeline(); //添加配置的handler即.handler(new LoggingHandler(LogLevel.INFO)) ChannelHandler handler config.handler(); if (handler ! null) { pipeline.addLast(handler); } //这里还没有和具体的eventLoop绑定 //需要等待NioServerSocketChannel注册完成 //即reigister方法调用完成才拥有eventLoop //execute方法是往taskQueue.offer(task); ch.eventLoop().execute(new Runnable() { Override public void run() { //构造ServerBootstrapAcceptor //负责接收客户端连接对连接的初始化工作。 //主要是接收到连接以后 为连接设置 option hanlder 设置属性 //currentChildHandler对应的是 wokrGroup的.childHandler()方法 pipeline.addLast(new ServerBootstrapAcceptor( ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } }); } init() 方法的源码比较长我们依然拆解成两个部分来看 1.设置OPTION参数 第一步设置 Socket 参数以及用户自定义属性。在创建服务端 Channel 时Channel 的配置参数保存在 NioServerSocketChannelConfig 中在初始化 Channel 的过程中Netty 会将这些参数设置到 JDK 底层的 Socket 上并把用户自定义的属性绑定在 Channel 上。 java public NioServerSocketChannel(ServerSocketChannel channel) { super(null, channel, SelectionKey.OP_ACCEPT); //创建channel时已经创建了config config new NioServerSocketChannelConfig(this, javaChannel().socket()); } 2.添加匿名ChannelInitializer java Override void init(Channel channel) { setChannelOptions (channel, options0().entrySet().toArray(newOptionArray(0)), logger); setAttributes(channel, attrs0().entrySet().toArray(newAttrArray(0))); ChannelPipeline p channel.pipeline();final EventLoopGroup currentChildGroup childGroup;final ChannelHandler currentChildHandler childHandler;final EntryChannelOption?, Object[] currentChildOptions childOptions.entrySet().toArray(newOptionArray(0));final EntryAttributeKey?, Object[] currentChildAttrs childAttrs.entrySet().toArray(newAttrArray(0));//ChannelInitializer一次性、初始化handler://负责添加一个ServerBootstrapAcceptor handler添加完后自己就移除了://ServerBootstrapAcceptor handler 负责接收客户端连接创建连接后对连接的初始化工作。p.addLast(new ChannelInitializerChannel() {Overridepublic void initChannel(final Channel ch) {//获取pipelinefinal ChannelPipeline pipeline ch.pipeline();//获取handler方法中指定的handlerChannelHandler handler config.handler();//如果handler不为空 加入 pipelineif (handler ! null) {pipeline.addLast(handler);}//添加1个任务//任务添加了ServerBootstrapAcceptor//在ServerBootstrapAcceptor中会将childHandler添加到piplinech.eventLoop().execute(new Runnable() {Overridepublic void run() {pipeline.addLast(new ServerBootstrapAcceptor(ch, currentChildGroup,currentChildHandler, currentChildOptions,currentChildAttrs));}});}});
} 第二步:添加匿名ChannelInitializer。 1.在ServerBootstrap#init为 Pipeline 添加了一个 匿名ChannelInitializerChannelInitializer 是实现了 ChannelHandler 接口的匿名类其中 ChannelInitializer 实现的 initChannel() 方法做了2件事: 1.将handler方法中指定的处理器加入到pipeline 2.通过task 的方式又向 Pipeline添加了一个处理器ServerBootstrapAcceptor。 从 ServerBootstrapAcceptor 的命名可以看出这是一个连接接入器专门用于接收新的连接然后把事件分发给 EventLoop 执行在这里我们先不做展开。 此时服务端的 pipeline 内部结构发生了变化如下图所示。 思考一个问题为什么需要 ChannelInitializer 处理器呢 ServerBootstrapAcceptor 的注册过程为什么又需要封装成异步 task 呢 因为我们在初始化时还没有将 Channel 注册到 Selector 对象上所以还无法注册 Accept 事件到 Selector 上(只是eventLoop和Selector 做了绑定,selecor在创建eventLoop的时候调用openSelector方法创建的) 所以事先在pipeline添加了 ChannelInitializer 处理器等待 Channel 注册完成后再向 Pipeline 中添加 ServerBootstrapAcceptor 处理器。 服务端 Channel 初始化的过程已经结束了。 整体流程比较简单主要是设置 Socket 参数以及用户自定义属性并向 Pipeline 中添加了1个特殊的处理器。 接下来我们继续分析如何将初始化好的 Channel 注册到 Selector 对象上 注册服务端 Channel 回到 initAndRegister() 的主流程创建完服务端 Channel 之后继续一层层跟进 register() 方法的源码 这里的逻辑是由bossGroup中的eventLoop处理。 java // MultithreadEventLoopGroup#register public ChannelFuture register(Channel channel) { // 选择一个eventLoop注册 //next(): //return executors[Math.abs(idx.getAndIncrement() % executors.length)]; //其实就是选择一个eventLoop将channel注册上去 //注意eventLoop已经和selector绑定了 return next().register(channel); } //SingleThreadEventLoop#register public ChannelFuture register(Channel channel) { //注意这里的channel被封装进入了一个 DefaultChannelPromise //this是MultithreadEventLoop return register(new DefaultChannelPromise(channel, this)); } //SingleThreadEventLoop#register public ChannelFuture register(final ChannelPromise promise) { ObjectUtil.checkNotNull(promise, promise); this是SingleThreadEventLoop promise.channel().unsafe().register(this, promise); return promise; } // AbstractChannel#register public final void register(EventLoop eventLoop, final ChannelPromise promise) { // 省略其他代码 AbstractChannel.this.eventLoop eventLoop; // 判断当前eventLoop是否是本线程调用 // boosGroup线程组线程内部调用 if (eventLoop.inEventLoop()) { register0(promise); } else { // 外部线程调用 try { eventLoop.execute(new Runnable() { Override public void run() { register0(promise); } }); } catch (Throwable t) { // 省略其他代码 } } } Netty 会在线程池 BossEventLoopGroup 中选择一个 EventLoop 与当前 Channel 进行绑定之后 Channel 生命周期内的所有 I/O 事件都由这个 EventLoop 负责处理如 accept、connect、read、write 等 I/O 事件。 可以看出不管是 EventLoop 线程本身调用还是外部线程用最终都会通过 register0() 方法进行注册 java private void register0(ChannelPromise promise) { try { if (!promise.setUncancellable() || !ensureOpen(promise)) { return; } boolean firstRegistration neverRegistered; // 调用 JDK 底层的 register() 进行注册 doRegister(); neverRegistered false; registered true; //触发匿名处理器的handlerAdded事件 //底层调用了callHandlerAdded0 //并调用了匿名处理器地initChannel方法 // 1.向pipeline添加handler中指定的处理器 // 2.向pipeline添加ServerBootStrapAcceptor //注意这2个是针对服务器端的channel添加的处理器。 //然后接收到新连接以后会使用这2个处理器处理客户端channel //ServerBootStrapAcceptor会把childHandler方法中的处理器放到客户端channel的pipeline pipeline.invokeHandlerAddedIfNeeded(); safeSetSuccess(promise); // 触发 channelRegistered 事件 pipeline.fireChannelRegistered(); // 此时 Channel 还未注册绑定地址所以处于非活跃状态 if (isActive()) { if (firstRegistration) { // Channel 当前状态为活跃时触发 channelActive 事件 pipeline.fireChannelActive(); } else if (config().isAutoRead()) { beginRead(); } } } catch (Throwable t) { // 省略其他代码 } } register0() 主要做了四件事 java 1.调用 JDK 底层进行 Channel 注册、 2.触发 handlerAdded 事件、 3.触发 channelRegistered 事件、 4.Channel 当前状态为活跃时触发 channelActive 事件。 1.注册Channel到选择器和注册事件0 为什么注册0因为还没初始化完成 我们对它们逐一进行分析。 首先看下 JDK 底层注册 Channel 的过程对应 doRegister() 方法的实现逻辑。 java protected void doRegister() throws Exception { boolean selected false; for (;;) { try { // 调用 JDK 底层的 register() 进行注册 // eventLoop().unwrappedSelector()指的是未包装的selector // 包装的selector指的是 selectKey // 注意这里注册的事件是 0 是 0 是 0 // 注意这里注册的事件是 0 是 0 是 0 // 注意这里注册的事件是 0 是 0 是 0 // this NioServerSocketChannel selectionKey javaChannel().register (eventLoop().unwrappedSelector(), 0, this); return; } catch (CancelledKeyException e) { // 省略其他代码 } } } public final SelectionKey register(Selector sel, int ops, Object att)throws ClosedChannelException{ synchronized (regLock) { // 这里是查询1个选择器并返回 SelectionKey k findKey(sel); if (k ! null) { //注册事件 0 k.interestOps(ops); //设置附件 //att NioServerSocketChannel k.attach(att); } if (k null) { synchronized (keyLock) { if (!isOpen()) throw new ClosedChannelException(); k ((AbstractSelector)sel).register(this, ops, att); addKey(k); } } return k; } } javaChannel().register() 负责调用 JDK 底层将 Channel 注册到 Selector 上register() 的第三个入参传入的是 Netty 自己实现的 NioServerSocketChannel 对象调用 register() 方法会将NioServerSocketChannel 绑定在 JDK 底层 Channel 的 attachment 上。 这样在每次 Selector对象进行事件循环时,Netty都可以从返回的JDK底层Channel中获得自己的Channel对象。 2.第一次注册触发handlerAdded事件 完成 Channel 向 Selector 注册后接下来就会触发 Pipeline 一系列的事件传播。 在事件传播之前用户自定义的业务处理器是如何被添加到 Pipeline 中的呢 答案:就在pipeline.invokeHandlerAddedIfNeeded() 当中。 java final void invokeHandlerAddedIfNeeded() { assert channel.eventLoop().inEventLoop(); //只有第一次注册才会调用该方法 否则不会调用该方法 if (firstRegistration) { firstRegistration false; System.out.println(invokeHandlerAddedIfNeeded:添加处理器); callHandlerAddedForAllHandlers(); } } 我们重点看下 handlerAdded 事件的处理过程。invokeHandlerAddedIfNeeded() 方法的调用层次比较深推荐你结合上述 Echo 服务端示例使用 IDE Debug 的方式跟踪调用栈如下图所示。 java private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) { try { ctx.setAddComplete(); //ctx.handler()只有1个就是我们初始化的时候添加的匿名ChannelInitializer //调用对应ChannelInitializer的handlerAdded ctx.handler().handlerAdded(ctx); } catch (Throwable t) { boolean removed false; try { remove0(ctx); try { ctx.handler().handlerRemoved(ctx); } finally { ctx.setRemoved(); } removed true; } catch (Throwable t2) { if (logger.isWarnEnabled()) { logger.warn(Failed to remove a handler: ctx.name(), t2); } } if (removed) {fireExceptionCaught(new ChannelPipelineException(ctx.handler().getClass().getName() .handlerAdded() has thrown an exception; removed., t));} else {fireExceptionCaught(new ChannelPipelineException(ctx.handler().getClass().getName() .handlerAdded() has thrown an exception; also failed to remove., t));}}
} 我们首先抓住 ChannelInitializer 中的核心源码逐层进行分析。 java // ChannelInitializer public void handlerAdded(ChannelHandlerContext ctx) throws Exception { if (ctx.channel().isRegistered()) { //调用初始化方法 if (initChannel(ctx)) { removeState(ctx); } } } //ChannelInitializer private boolean initChannel(ChannelHandlerContext ctx) throws Exception { if (initMap.add(ctx)) { try { //调用匿名ChannelInitializer重写的initChannel()方法 //1.向 Pipeline 添加 handler方法中指定的handler //2.通过异步任务添加ServerBootstrapAcceptor处理器 initChannel((C) ctx.channel()); } catch (Throwable cause) { exceptionCaught(ctx, cause); } finally { ChannelPipeline pipeline ctx.pipeline(); if (pipeline.context(this) ! null) { // 将 ChannelInitializer 自身从 Pipeline 中移出 pipeline.remove(this); } } return true; } return false; } 可以看出 ChannelInitializer 首先会调用 initChannel() 抽象方法。 然后 Netty 会把 ChannelInitializer 自身从 Pipeline移除。 其中 initChannel() 抽象方法是在哪里实现的呢 这就要跟踪到 ServerBootstrap 的 init() 方法其中有这么一段代码 java //此处的channel是服务器端的channel p.addLast(new ChannelInitializerChannel() { Override public void initChannel(final Channel ch) { final ChannelPipeline pipeline ch.pipeline //添加的是handler方法中指定的处理器。 //即服务器端配置的handler方法中指定的处理器。 //handler方法中指定的处理器在初始化时就会执行 //而childHandler方法中指定的处理器会在客户端成功connect后才执行这是两者的区别。 ChannelHandler handler config.handler(); if (handler ! null) { pipeline.addLast(handler); } ch.eventLoop().execute(new Runnable() { Override public void run() { //handler方法中指定的处理器在初始化时就会执行 //而childHandler方法中指定的处理器会在客户端成功connect后才执行这是两者的区别。 //ServerBootstrapAcceptor负责接收客户端连接。 //创建客户端连接后对连接的初始化工作。 //因为添加 ServerBootstrapAcceptor是一个异步过程需要EventLoop线程负责执行。 //而当前 EventLoop线程正在执行 register0() 的注册流程 //所以等到 register0() 执行完之后才能被添加到Pipeline当中。 //此时已经注册完成 pipeline.addLast(new ServerBootstrapAcceptor( ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } }); 3.调用匿名ChannelInitializer#initChannel 4.向服务器channel添加handler()中的handler java ChannelHandler handler config.handler(); if (handler ! null) { System.out.println (向服务器端添加serverBootstrap.handler() 方法中指定的处理器ServerLoggingHandler); pipeline.addLast(handler); } 5.触发LoggingHandler#handlerAdded 此时会触发处理器的handlerAdded事件,调用ChannelHandlerAdapter#handlerAdded的方法。 java final void callHandlerAdded() throws Exception { if (setAddComplete()) { //handler是当前被添加的处理器ServerLoggingHandler //ServerLoggingHandler继承了ChannelDuplexHandler //ChannelDuplexHandler extends ChannelInboundHandlerAdapter //所以调用的是ChannelInboundHandlerAdapter#handlerAdded handler().handlerAdded(this); } } io.netty.channel.ChannelHandlerAdapter#handlerAdded java Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { //匿名ChannelInitializer继承自ChannelHandlerAdapter //在调用handlerAdded时会触发ChannelHandlerAdapter的handlerAdded方法 System.out.println(触发 this.getClass().getSimpleName() .handlerAdded); } 从代码的走势看应该添加 ServerBootstrapAcceptor 。 但是因为添加 ServerBootstrapAcceptor 是一个异步过程需要 EventLoop 线程负责执行。 而当前 EventLoop 线程正在执行 register0() 的注册流程所以等到 register0() 执行完之后才能被添加到 Pipeline 当中。也就是说添加ServerBootstrapAcceptor的runnable正在队列中等待被执行。 java 1.当前线程正在执行register0代码 2.当前线程代码中执行initChannel 3.当前线程然后添加handler方法中的处理器 4.当前线程中执行以下代码 ch.eventLoop().execute(new Runnable() { //添加SBAcceptor } 此时只是给线程添加了1个任务,然后放入了队列中 只有当当前线程把正在执行的register0代码执行完毕 才能继续把队列中的任务取出来执行 java package io.netty.example.echo; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class NettyThread { public static void main(String[] args) { final ExecutorService executorService Executors.newFixedThreadPool(1); final Runnable runnable2 new Runnable() { Override public void run() { System.out.println(2); } }; final Runnable runnable1 new Runnable() {Overridepublic void run() {System.out.println(我现在在执行register0开始);executorService.execute(runnable2);System.out.println(我现在在执行register0结束);}};executorService.execute(runnable1);
} } /* 我现在在执行register0开始 我现在在执行register0结束 2 */ 完成 initChannel() 这一步之后ServerBootstrapAcceptor 并没有被添加到 Pipeline 中此时 Pipeline 的内部结构变化如下图所示。 Head --- LoggingHandler --- Tail 6.触发LoggingHandler的channelRegistered 我们回到 register0() 的主流程接着向下分析。 channelRegistered 事件是由 fireChannelRegistered() 方法触发沿着 Pipeline 的 Head 节点传播到 Tail 节点并依次调用每个 ChannelHandler 的 channelRegistered() 方法。此时pipeline的示意图如下 Head --- LoggingHandler --- Tail 所以只会触发LoggingHandler的channelRegistered java 触发事件:fireChannelRegistered ServerLoggingHandler.channelRegistered 然而此时 Channel 还未注册绑定地址所以处于非活跃状态所以并不会触发 channelActive 事件。 7. register0()的注册流程执行完毕 执行完整个 register0() 的注册流程之后。 EventLoop线程会将 ServerBootstrapAcceptor 添加到 Pipeline 当中。 java ch.eventLoop().execute(new Runnable() { Override public void run() { System.out.println(服务器端异步添加ServerBootstrapAcceptor); pipeline.addLast(new ServerBootstrapAcceptor( ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); 此时会触发处理器的handlerAdded事件,调用ChannelHandlerAdapter#handlerAdded的方法。 java final void callHandlerAdded() throws Exception { if (setAddComplete()) { //handler是当前被添加的处理器ServerLoggingHandler //ServerLoggingHandler继承了ChannelDuplexHandler //ChannelDuplexHandler extends ChannelInboundHandlerAdapter //所以调用的是ChannelInboundHandlerAdapter#handlerAdded handler().handlerAdded(this); } } io.netty.channel.ChannelHandlerAdapter#handlerAdded java Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { //匿名ChannelInitializer继承自ChannelHandlerAdapter //在调用handlerAdded时会触发ChannelHandlerAdapter的handlerAdded方法 System.out.println(触发 this.getClass().getSimpleName() .handlerAdded); } 此时 Pipeline 的内部结构又发生了变化如下图所示。 8.向服务器channel添加SBAcceptor 在前面我们已经分析了 initChannel() 方法的实现逻辑首先向 Pipeline 中添加 ServerSocketChannel 对应的 Handler然后通过异步 task 的方式向 Pipeline 添加 ServerBootstrapAcceptor 处理器。 其中有一个点不要混淆handler() 方法中的处理器和 ServerBootstrapAcceptor 处理器是添加到服务端的Pipeline 上。 而 childHandler() 方法是通过 ServerBootstrapAcceptor 处理器添加到客户端的 Pipeline 上。 9.端口绑定 整个服务端 Channel 注册的流程我们已经讲完注册过程中 Pipeline 结构的变化值得你再反复梳理从而加深理解。 目前服务端还是不能工作的还差最后一步就是进行端口绑定我们继续向下分析。 回到 ServerBootstrap 的 bind() 方法我们继续跟进端口绑定 doBind0() 的源码。 java public final void bind(final SocketAddress localAddress, final ChannelPromise promise) { assertEventLoop(); // 省略其他代码 boolean wasActive isActive(); try { // 调用 JDK 底层进行端口绑定 doBind(localAddress); } catch (Throwable t) { safeSetFailure(promise, t); closeIfClosed(); return; } if (!wasActive isActive()) { invokeLater(new Runnable() { Override public void run() { // 触发 channelActive 给ServerSocketChannel注册 // SelectionKey.OP_ACCEPT事件 // 所有事件的触发都是通过pipeline pipeline.fireChannelActive(); } }); } safeSetSuccess(promise); } bind() 方法主要做了两件事: 1.调用 JDK 底层进行端口绑定 2.绑定成功后并触发 channelActive 事件。下面我们逐一进行分析。 首先看下调用 JDK 底层进行端口绑定的 doBind() 方法 java protected 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 真正已经完成端口绑定了。 10.触发 channelActive 事件 完成端口绑定之后Channel 处于活跃 Active 状态然后会调用 pipeline.fireChannelActive() 方法触发 channelActive 事件。 即Channel 处于就绪状态可以被读写。 我们可以一层层跟进 fireChannelActive() 方法发现其中比较重要的部分 java // DefaultChannelPipeline#channelActive public void channelActive(ChannelHandlerContext ctx) { ctx.fireChannelActive(); readIfIsAutoRead(); } // AbstractNioChannel#doBeginRead protected void doBeginRead() throws Exception { // Channel.read() or ChannelHandlerContext.read() was called final SelectionKey selectionKey this.selectionKey; if (!selectionKey.isValid()) { return; } readPending true; final int interestOps selectionKey.interestOps(); if ((interestOps readInterestOp) 0) { // 注册 OP_ACCEPT 事件到服务端 Channel 的事件集合 selectionKey.interestOps(interestOps | readInterestOp); } } 可以看出在执行完 channelActive 事件传播之后会调用 readIfIsAutoRead() 方法触发 Channel 的 read 事件而它最终调用到 AbstractNioChannel 中的 doBeginRead() 方法其中 readInterestOp 参数就是在前面初始化 Channel 所传入的 SelectionKey.OPACCEPT 事件所以 OPACCEPT 事件会被注册到 Channel 的事件集合中。 11.监听Accept事件:16 java Override protected void doBeginRead() throws Exception { // Channel.read() or ChannelHandlerContext.read() was called final SelectionKey selectionKey this.selectionKey; if (!selectionKey.isValid()) { return; } readPending true; final int interestOps selectionKey.interestOps(); //假设之前没有监听readInterestOp则监听readInterestOp if ((interestOps readInterestOp) 0) { //NioServerSocketChannel: readInterestOp OP_ACCEPT 1 4 16 logger.info(interest ops readInterestOp); selectionKey.interestOps(interestOps | readInterestOp); } } 到此为止整个服务端已经真正启动完毕。我们总结一下服务端启动的全流程如下图所示。 创建服务端 Channel本质是创建 JDK 底层原生的 Channel并初始化几个重要的属性包括 id、unsafe、pipeline 等。 初始化服务端 Channel设置 Socket 参数以及用户自定义属性并添加1个特殊的处理器 ChannelInitializer,ChannelInitializer的功能是添加 LoggingHandler 和 ServerBootstrapAcceptor,但是并没有添加进去。 注册服务端 Channel调用 JDK 底层将 Channel 注册到 Selector上。执行ChannelInitializer的initChannel真正添加handler。 端口绑定调用 JDK 底层进行端口绑定并触发 channelActive 事件把 OP_ACCEPT 事件注册到NioServerSocketChannel 的事件集合中。 总结 java •启动服务的本质 Selector selector sun.nio.ch.SelectorProviderImpl.openSelector() ServerSocketChannel serverSocketChannel provider.openServerSocketChannel() selectionKey javaChannel().register(eventLoop().unwrappedSelector(), 0, this); javaChannel().bind(localAddress, config.getBacklog()); selectionKey.interestOps(OP_ACCEPT); 知识点: •Selector 是在 new NioEventLoopGroup()(创建一批 NioEventLoop)时创建。 •第一次 Register 并不是监听 OPACCEPT而是 0: selectionKey javaChannel().register(eventLoop().unwrappedSelector(), 0, this) 。 •最终监听 OPACCEPT 是通过 bind 完成后的 fireChannelActive() 来触发的。 •NioEventLoop 是通过 Register 操作的执行来完成启动的。 •类似 ChannelInitializer一些 Hander 可以设计成一次性的用完就移除例如授权
