1. 提供了一个ChannelFactory对象用来创建Channel,一个Channel会对应一个EventLoop用于IO的事件处理，在一个Channel的整个生命周期中 只会绑定一个EventLoop,这里可理解给Channel分配一个线程进行IO事件处理，结束后回收该线程。
2. AbstractBootstrap没有提供EventLoop而是提供了一个EventLoopGroup，其实我认为这里只用一个EventLoop就行了。
3. 不管是服务器还是客户端的Channel都需要绑定一个本地端口这就有了SocketAddress类的对象localAddress。
4. Channel有很多选项所有有了options对象LinkedHashMap, Object>
5. 怎么处理Channel的IO事件呢，我们添加一个事件处理器ChannelHandler对象。

• EventLoop初始化
• channel的注册
• pipeline的初始化
• handler的添加过程
• 服务端连接处理。

public class ServerBootstrap extends AbstractBootstrap {private static final InternalLogger logger = InternalLoggerFactory.getInstance(ServerBootstrap.class); // The order in which child ChannelOptions are applied is important they may depend on each other for validation // purposes. //SocketChannel相关的属性配置 private final Map, Object> childOptions = new LinkedHashMap, Object>(); private final Map, Object> childAttrs = new ConcurrentHashMap, Object>(); private final ServerBootstrapConfig config = new ServerBootstrapConfig(this); //配置类 private volatile EventLoopGroup childGroup; //工作线程组 private volatile ChannelHandler childHandler; //负责SocketChannel的IO处理相关的Handlerpublic ServerBootstrap() { } }

• Netty自己实现的Channel与底层JDK提供的Channel是如何联系并且构建实现的
• ChannelInitializer这个特殊的Handler处理器的作用以及实现原理
• Pipeline是如何初始化以的

ChannelFuture channelFuture=bootstrap.bind(port).sync();

public ChannelFuture bind(SocketAddress localAddress) { validate(); return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress")); }

• validate()， 验证ServerBootstrap核心成员属性的配置是否正确，比如group、channelFactory、childHandler、childGroup等，这些属性如果没配置，那么服务端启动会报错
• localAddress，绑定一个本地端口地址

• initAndRegister 初始化并注册Channel，并返回一个ChannelFuture，说明初始化注册Channel是异步实现
• regFuture.cause() 用来判断initAndRegister()是否发生异常，如果发生异常，则直接返回
• regFuture.isDone()， 判断initAndRegister()方法是否执行完成。
  
  • 如果执行完成，则调用doBind0()方法。
  • 如果未执行完成，regFuture添加一个监听回调，在监听回调中再次判断执行结果进行相关处理。
  • PendingRegistrationPromise 用来保存异步执行结果的状态

private ChannelFuture doBind(final SocketAddress localAddress) { final ChannelFuture regFuture = initAndRegister(); final Channel channel = regFuture.channel(); if (regFuture.cause() != null) { return regFuture; }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; } }

• 初始化，应该就是构建服务端的Handler处理链
• register，应该就是把当前服务端的连接注册到selector上

final ChannelFuture initAndRegister() { Channel channel = null; try { //通过ChannelFactory创建一个具体的Channel实现 channel = channelFactory.newChannel(); init(channel); //初始化 } catch (Throwable t) { //省略.... } //这个代码应该和我们猜想是一致的，就是将当前初始化的channel注册到selector上，这个过程同样也是异步的 ChannelFuture regFuture = config().group().register(channel); if (regFuture.cause() != null) { //获取regFuture的执行结果 if (channel.isRegistered()) { channel.close(); } else { channel.unsafe().closeForcibly(); } } return regFuture; }

bootstrap.group(bossGroup, workerGroup) //配置Server的通道，相当于NIO中的ServerSocketChannel .channel(NioServerSocketChannel.class)

public B channel(Class channelClass) { return channelFactory(new ReflectiveChannelFactory( ObjectUtil.checkNotNull(channelClass, "channelClass") )); }

@Override public T newChannel() { try { return constructor.newInstance(); } catch (Throwable t) { throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t); } }

而此时的clazz其实就是：NioServerSocketChannel

public class ReflectiveChannelFactory implements ChannelFactory {private final Constructor constructor; public ReflectiveChannelFactory(Class 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); } } }

public class NioServerSocketChannel extends AbstractNioMessageChannel implements io.netty.channel.socket.ServerSocketChannel { private static ServerSocketChannel newSocket(SelectorProvider provider) { try { return provider.openServerSocketChannel(); } catch (IOException e) { throw new ChannelException( "Failed to open a server socket.", e); } } public NioServerSocketChannel() { this(newSocket(DEFAULT_SELECTOR_PROVIDER)); } }

public ServerSocketChannel openServerSocketChannel() throws IOException { return new ServerSocketChannelImpl(this); }

public NioServerSocketChannel(ServerSocketChannel channel) { super(null, channel, SelectionKey.OP_ACCEPT); config = new NioServerSocketChannelConfig(this, javaChannel().socket()); }

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) { super(parent); this.ch = ch; this.readInterestOp = readInterestOp; //设置关心事件，此时是一个连接事件，所以是OP_ACCEPT try { 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); } }

final ChannelFuture initAndRegister() { //省略.... //这个代码应该和我们猜想是一致的，就是将当前初始化的channel注册到selector上，这个过程同样也是异步的 ChannelFuture regFuture = config().group().register(channel); if (regFuture.cause() != null) { //获取regFuture的执行结果 if (channel.isRegistered()) { channel.close(); } else { channel.unsafe().closeForcibly(); } } return regFuture; }

@Override public ChannelFuture register(Channel channel) { return next().register(channel); }

@Override public ChannelFuture register(Channel channel) { return register(new DefaultChannelPromise(channel, this)); }

@Override public ChannelFuture register(final ChannelPromise promise) { ObjectUtil.checkNotNull(promise, "promise"); promise.channel().unsafe().register(this, promise); return promise; }

@Override public 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)) { //判断eventLoop类型是否是EventLoop对象类型，如果不是则抛出异常 promise.setFailure( new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName())); return; }AbstractChannel.this.eventLoop = eventLoop; //Reactor内部线程调用，也就是说当前register方法是EventLoop线程触发的，则执行下面流程 if (eventLoop.inEventLoop()) { register0(promise); } else { //如果是外部线程 try { eventLoop.execute(new Runnable() { @Override public 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); } } }

• 调用JDK层面的API对当前Channel进行注册
• 触发HandlerAdded事件
• 触发channelRegistered事件
• Channel状态为活跃时，触发channelActive事件

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; doRegister(); //调用JDK层面的register()方法进行注册 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(); //触发Handler，如果有必要的情况下safeSetSuccess(promise); 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()) { //此时是ServerSocketChannel的注册，所以连接还处于非活跃状态 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/4805 beginRead(); } } } catch (Throwable t) { // Close the channel directly to avoid FD leak. closeForcibly(); closeFuture.setClosed(); safeSetFailure(promise, t); } }

@Override protected 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; } } } }

• 初始化指定的Channel实例
• 把该Channel分配给某一个EventLoop
• 然后把Channel注册到该EventLoop的Selector中

private static void doBind0( final ChannelFuture regFuture, final Channel channel, final SocketAddress localAddress, final ChannelPromise promise) {//获取当前channel中的eventLoop实例，执行一个异步任务。 //需要注意，以前我们在课程中讲过，eventLoop在轮询中一方面要执行select遍历，另一方面要执行阻塞队列中的任务，而这里就是把任务添加到队列中异步执行。 channel.eventLoop().execute(new Runnable() { @Override public void run() { //如果ServerSocketChannel注册成功，则调用该channel的bind方法 if (regFuture.isSuccess()) { channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE); } else { promise.setFailure(regFuture.cause()); } } }); }

protected void doBind(SocketAddress localAddress) throws Exception { if (PlatformDependent.javaVersion() >= 7) { javaChannel().bind(localAddress, config.getBacklog()); } else { javaChannel().socket().bind(localAddress, config.getBacklog()); } }

.childHandler(new ChannelInitializer() { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { socketChannel.pipeline() .addLast(new NormalInBoundHandler("NormalInBoundA",false)) .addLast(new NormalInBoundHandler("NormalInBoundB",false)) .addLast(new NormalInBoundHandler("NormalInBoundC",true)); socketChannel.pipeline() .addLast(new NormalOutBoundHandler("NormalOutBoundA")) .addLast(new NormalOutBoundHandler("NormalOutBoundB")) .addLast(new NormalOutBoundHandler("NormalOutBoundC")) .addLast(new ExceptionHandler()); } });

final ChannelFuture initAndRegister() { Channel channel = null; try { channel = channelFactory.newChannel(); //这是是创建channel init(channel); //这里是初始化 } catch (Throwable t) { //省略.... } ChannelFuture regFuture = config().group().register(channel); //这是是注册 if (regFuture.cause() != null) { if (channel.isRegistered()) { channel.close(); } else { channel.unsafe().closeForcibly(); } }return regFuture; }

@Override void init(Channel channel) { setChannelOptions(channel, newOptionsArray(), logger); setAttributes(channel, newAttributesArray()); ChannelPipeline p = channel.pipeline(); final EventLoopGroup currentChildGroup = childGroup; final ChannelHandler currentChildHandler = childHandler; //childHandler就是在服务端配置时添加的ChannelInitializer final Entry, Object>[] currentChildOptions = newOptionsArray(childOptions); final Entry, Object>[] currentChildAttrs = newAttributesArray(childAttrs); // 此时的Channel是NioServerSocketChannel，这里是为NioServerSocketChannel添加处理器链。 p.addLast(new ChannelInitializer() { @Override public void initChannel(final Channel ch) { final ChannelPipeline pipeline = ch.pipeline(); ChannelHandler handler = config.handler(); //如果在ServerBootstrap构建时，通过.handler添加了处理器，则会把相关处理器添加到NioServerSocketChannel中的pipeline中。 if (handler != null) { pipeline.addLast(handler); }ch.eventLoop().execute(new Runnable() { //异步天剑一个ServerBootstrapAcceptor处理器，从名字来看， @Override public void run() { pipeline.addLast(new ServerBootstrapAcceptor( //currentChildHandler，表示SocketChannel的pipeline，当收到客户端连接时，就会把该handler添加到当前SocketChannel的pipeline中 ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } }); }

• ChannelPipeline 是在AbstractChannel中的构造方法中初始化的一个DefaultChannelPipeline
  

  protected AbstractChannel(Channel parent) { this.parent = parent; id = newId(); unsafe = newUnsafe(); pipeline = newChannelPipeline(); }

  
  
• p.addLast是为NioServerSocketChannel添加handler处理器链，这里添加了一个ChannelInitializer回调函数，该回调是异步触发的，在回调方法中做了两件事
  
  • 如果ServerBootstrap.handler添加了处理器，则会把相关处理器添加到该pipeline中，在本次演示的案例中，我们添加了LoggerHandler
  • 异步执行添加了ServerBootstrapAcceptor，从名字来看，它是专门用来接收新的连接处理的。

1. 启动服务端监听
2. 在ServerBootstrapAcceptor的channelRead方法中打上断点
3. 通过telnet 连接，此时会触发debug。

public void channelRead(ChannelHandlerContext ctx, Object msg) { final Channel child = (Channel) msg; child.pipeline().addLast(childHandler); //在这里，将handler添加到SocketChannel的pipeline中setChannelOptions(child, childOptions, logger); setAttributes(child, childAttrs); try { //把当前客户端的链接SocketChannel注册到某个EventLoop中。 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); } }

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; 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); pipeline.fireChannelRegistered(); //执行pipeline中的ChannelRegistered()事件。 // 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) { 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); } }

@Override public final ChannelPipeline fireChannelRegistered() { AbstractChannelHandlerContext.invokeChannelRegistered(head); return this; }

• 如果当前的任务是在eventLoop中触发的，则直接调用invokeChannelRegistered
• 否则，异步执行invokeChannelRegistered。

static void invokeChannelRegistered(final AbstractChannelHandlerContext next) { EventExecutor executor = next.executor(); if (executor.inEventLoop()) { next.invokeChannelRegistered(); } else { executor.execute(new Runnable() { @Override public void run() { next.invokeChannelRegistered(); } }); } }

private void invokeChannelRegistered() { if (invokeHandler()) { try { ((ChannelInboundHandler) handler()).channelRegistered(this); } catch (Throwable t) { invokeExceptionCaught(t); } } else { fireChannelRegistered(); } }

• 创建服务端Channel，本质上是根据用户配置的实现，调用JDK原生的Channel
• 初始化Channel的核心属性，unsafe、pipeline
• 初始化Channel的Pipeline，主要是添加两个特殊的处理器，ChannelInitializer和ServerBootstrapAcceptor
• 注册服务端的Channel，添加OP_ACCEPT事件，这里底层调用的是JDK层面的实现，讲Channel注册到BossEventLoop中的Selector上
• 绑定端口，调用JDK层面的API，绑定端口。

版权声明：本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来自 Mic带你学架构！
如果本篇文章对您有帮助，还请帮忙点个关注和赞，您的坚持是我不断创作的动力。欢迎关注同名微信公众号获取更多技术干货！

推荐阅读

• 

  什么是碱性岩
• 

  洛阳奥斯卡影院6月13号放映什么电影？洛阳奥斯卡影院今日放映表
• 

  京东“京银计划” 是什么 怎么参与？
• 

  共栅极放大电路分析,场效应管共栅极放大电路
• 

  2022年三伏天时间表？
• 

  活珠子不适合哪些人吃
• 

  m当前“最接近完美”的五款手机，别看价格贵，用三年毫无压力
• 

  上海电商有哪些 什么是电商上海，什么是电商上海人
• 

  家用茶具多少钱的合适 – 高档茶盘图片价格表
• 

  冰箱不结冰是怎么回事？原因分析与解决方法
• 

  热水器不烧水怎么办,是因为这些原因
• 

  增加注册资本后股东的股权比例
• 

  白发傅筹第几集知道身世
• 

  CPB水前肌底精华好不好？CPB水前肌底精华怎么用
• 

  兄弟T310如何清零废墨？
• 

  炒杜仲与生杜仲的区别 炒杜仲与生杜仲的不同之处
• 

  《侠客风云传前传》激活方法及特权简单介绍
• 

  取暖器品牌排名？哪个比较好的？
• 

  中风|女性中风有奇特症状？盘点易被女性忽视的11个中风症状！
• 

  减肥可以吃红枣吗

• 慢慢的美丽
• 开学第一天（下）
• 奔向你的城市
• 学无止境，人生还很长
• 由浅入深理解AOP
• “成长”读书社群招募
• 继续努力，自主学习家庭Day135(20181015）
• python学习之|python学习之 实现QQ自动发送消息
• 每日一话（49）——一位清华教授在朋友圈给大学生的9条建议
• 小影写在2018九月开学季