netty源码解析系列

• Netty 源码解析系列-服务端启动流程解析
• Netty 源码解析系列-客户端连接接入及读I/O解析
• 五分钟就能看懂pipeline模型 -Netty 源码解析

1.服务端启动例子(基于4.0.31.Final)

   public class Server {private ServerBootstrap serverBootstrap;private NioEventLoopGroup bossGroup;private NioEventLoopGroup workGroup;public static void main(String[] args) throws InterruptedException {System.out.println("服务启动");Server server = new Server();server.start();}private void start() throws InterruptedException {try {serverBootstrap=new ServerBootstrap();bossGroup = new NioEventLoopGroup();workGroup = new NioEventLoopGroup(4);serverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT).childOption(ChannelOption.SO_KEEPALIVE, true).handler(new InitHandler()).childHandler(new IOChannelInitialize());ChannelFuture future = serverBootstrap.bind(8802).sync();future.channel().closeFuture().sync();} finally {bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}}private class IOChannelInitialize extends ChannelInitializer<SocketChannel>{@Overrideprotected void initChannel(SocketChannel ch) throws Exception {System.out.println("initChannel");ch.pipeline().addLast(new IdleStateHandler(1000, 0, 0));ch.pipeline().addLast(new IOHandler());}}
}
复制代码

步骤说明

• 1.1 创建 ServerBootstrap 实例，它是 netty 的启动辅助类，提供了一系列的方法用于设置服务 端启动相关的参数。底层通过门面模式对各种能力进行抽象和封装，尽量不需要用户跟过 多的底层 API 打交道，降低用户的开发难度

• 1.2 NioEventLoopGroup 是 netty Reactor 线程池，bossGroup 监听和 accept 客户端连接，workGroup 则处理 IO ，编解码

• 1.3 绑定服务端 NioServerSocketChannel

• 1.4 设置一些参数

• 1.5 初始化 pipeline 并绑定 handler ，pipeline 是一个负责处理网络事件的职责链，负责管理和执行 ChannelHandler ，设置系统提供的 IdleStateHandler 和自定义 IOHandler

• 1.6 serverBootstrap.bind(8802) 这里才是启动服务端绑定端口

• 1.7 future.channel().closeFuture().sync(); 等待服务端关闭

• 1.8 优雅关闭

2. 源码分析

2.1 NioEventLoopGroup

    NioEventLoopGroup 不仅仅是 I/O 线程，除了负责 I/O 的读写，还负责系统 Task 和定时任务

     public NioEventLoopGroup(int nThreads) {this(nThreads, null);}
复制代码

public NioEventLoopGroup(int nThreads, ThreadFactory threadFactory) {this(nThreads, threadFactory, SelectorProvider.provider());
}
复制代码

public NioEventLoopGroup(int nThreads, ThreadFactory threadFactory, final SelectorProvider selectorProvider) {super(nThreads, threadFactory, selectorProvider);
}
复制代码

protected MultithreadEventLoopGroup(int nThreads, ThreadFactory threadFactory, Object... args) {super(nThreads == 0? DEFAULT_EVENT_LOOP_THREADS : nThreads, threadFactory, args);
}
复制代码

    继续，以下是精简代码

protected MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory, Object... args) {...if (threadFactory == null) {threadFactory = newDefaultThreadFactory();}children = new SingleThreadEventExecutor[nThreads];if (isPowerOfTwo(children.length)) {chooser = new PowerOfTwoEventExecutorChooser();} else {chooser = new GenericEventExecutorChooser();}for (int i = 0; i < nThreads; i ++) {...children[i] = newChild(threadFactory, args);...}复制代码

     MultithreadEventExecutorGroup 实现了线程的创建和线程的选择，我们看看 newChild 方法( NioEventLoopGroup 类的方法)，newChild 实例化线程

@Override
protected EventExecutor newChild(ThreadFactory threadFactory, Object... args) throws Exception {return new NioEventLoop(this, threadFactory, (SelectorProvider) args[0]);
}
复制代码

    创建了一个 NioEventLoop

NioEventLoop(NioEventLoopGroup parent, ThreadFactory threadFactory, SelectorProvider selectorProvider) {super(parent, threadFactory, false);if (selectorProvider == null) {throw new NullPointerException("selectorProvider");}provider = selectorProvider;selector = openSelector();
}
复制代码

    跟着 super

protected SingleThreadEventLoop(EventLoopGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp) {super(parent, threadFactory, addTaskWakesUp);
}
复制代码

    代码有精简，继续

protected SingleThreadEventExecutor(EventExecutorGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp) {thread = threadFactory.newThread(new Runnable() {@Overridepublic void run() {SingleThreadEventExecutor.this.run();}}
});
复制代码

    在这里实例化了一个线程，并在 run 中调用 SingleThreadEventExecutor 的 run 方法，这个线程在哪里启动的呢，我们继续往下看
    总结:
          NioEventLoopGroup 实际就是 Reactor 线程池，负责调度和执行客户端的接入、网络读写事件的处理、用户自定义任务和定时任务的执行。

2.2 ServerBootstrap

     ServerBootstrap 是服务端的启动辅助类，父类是 AbstractBootstrap ，与之相对应的客户端启动辅助类是 Bootstrap

    public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {volatile EventLoopGroup group;private volatile ChannelFactory<? extends C> channelFactory;private volatile SocketAddress localAddress;private final Map<ChannelOption<?>, Object> options = new LinkedHashMap<ChannelOption<?>, Object>();private final Map<AttributeKey<?>, Object> attrs = new LinkedHashMap<AttributeKey<?>, Object>();private volatile ChannelHandler handler;}
复制代码

2.2.1 设置booss和work线程池

     将 bossGroup 传给父类，workGroup 赋值给 serverBootstrap 的 childGroup

      public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {super.group(parentGroup);if (childGroup == null) {throw new NullPointerException("childGroup");}if (this.childGroup != null) {throw new IllegalStateException("childGroup set already");}this.childGroup = childGroup;return this;
}
复制代码

2.2.2 设置NioServerSocketChannel处理连接请求

serverBootstrap.channel(NioServerSocketChannel.class)
复制代码

   public B channel(Class<? extends C> channelClass) {if (channelClass == null) {throw new NullPointerException("channelClass");}return channelFactory(new BootstrapChannelFactory<C>(channelClass));}
复制代码

     继续跟 new BootstrapChannelFactory

      private static final class BootstrapChannelFactory<T extends Channel> implements ChannelFactory<T> {private final Class<? extends T> clazz;BootstrapChannelFactory(Class<? extends T> clazz) {this.clazz = clazz;}@Overridepublic T newChannel() {try {return clazz.newInstance();} catch (Throwable t) {throw new ChannelException("Unable to create Channel from class " + clazz, t);}}}
复制代码

     BootstrapChannelFactory 是一个继承了 ChannelFactory 的内部类，从名称上就能看出，这是一个 channel 工厂类，重写了父类的 newChannel 方法，通过反射创建 NioServerSocketChannel 实例，后面会告诉你是在哪里调用到的

2.2.3 设置channel通道块的值

         serverBootstrap.option(ChannelOption.SO_BACKLOG, 128)
复制代码

          public <T> B option(ChannelOption<T> option, T value) {if (option == null) {throw new NullPointerException("option");}if (value == null) {synchronized (options) {options.remove(option);}} else {synchronized (options) {options.put(option, value);}}return (B) this;}
复制代码

     这里的 option 方法是父类 AbstractBootstrap 的方法，options 是一个有序的非线程安全的双向链表，加锁添加

2.2.4 serverBootstrap.childOption

        public <T> ServerBootstrap childOption(ChannelOption<T> childOption, T value) {if (childOption == null) {throw new NullPointerException("childOption");}if (value == null) {synchronized (childOptions) {childOptions.remove(childOption);}} else {synchronized (childOptions) {childOptions.put(childOption, value);}}return this;}
复制代码

     childOption 是子类 serverBootstrap 的方法
     childOption 和 option 的区别:
         option : 主要是设置 ServerChannel 的一些选项
         childOption : 主要设置 ServerChannel 的子 channel 的选项，即 option
                       针对的是 boss 线程而 childOption 针对的是 work 线程池

2.2.5 设置服务端NioServerSocketChannel的Handler

         serverBootstrap.handler(new InitHandler())
复制代码

         public B handler(ChannelHandler handler) {if (handler == null) {throw new NullPointerException("handler");}this.handler = handler;return (B) this;}
复制代码

2.2.6 serverBootstrap.childHandler()

           public ServerBootstrap childHandler(ChannelHandler childHandler) {if (childHandler == null) {throw new NullPointerException("childHandler");}this.childHandler = childHandler;return this;}
复制代码

handler 和 childHandler 的区别
    Handler 是属于服务端 NioServerSocketChannel ，只会创建一次 childHandler 是属于每一个新建的 NioSocketChannel ，每当有一个连接上来，都会调用

2.2.7 真正的启动过程是在这里执行，我们看看bind()方法

         serverBootstrap.bind(8802).sync()
复制代码

         public ChannelFuture bind(int inetPort) {return bind(new InetSocketAddress(inetPort));}
复制代码

• (1) 通过端口号创建一个 InetSocketAddress ，继续 bind

 public ChannelFuture bind(SocketAddress localAddress) {validate();if (localAddress == null) {throw new NullPointerException("localAddress");}return doBind(localAddress);}
复制代码

• (2) validate() 方法进行一些参数验证，我们直接看 doBind()

        private ChannelFuture doBind(final SocketAddress localAddress) {final ChannelFuture regFuture = initAndRegister();final Channel channel = regFuture.channel();if (regFuture.cause() != null) {return regFuture;}if (regFuture.isDone()) {ChannelPromise promise = channel.newPromise();doBind0(regFuture, channel, localAddress, promise);return promise;} else {final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);regFuture.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {Throwable cause = future.cause();if (cause != null) {promise.setFailure(cause);} else {promise.executor = channel.eventLoop();}doBind0(regFuture, channel, localAddress, promise);}});return promise;}}
复制代码

• (3.1) 先看 initAndRegister ( AbstractBootstrap 类 )，去掉了一些不重要的

            final ChannelFuture initAndRegister() {final Channel channel = channelFactory().newChannel();init(channel);ChannelFuture regFuture = group().register(channel);return regFuture;}
复制代码

     channelFactory 是 serverBootstrap.channel() 时创建的，在这里调用反射创建 NioServerSocketChannel 实例

• (3.2.1) 再看 init(channel) 方法( ServerBootstrap 类)

    @Override
void init(Channel channel) throws Exception {final Map<ChannelOption<?>, Object> options = options();synchronized (options) {channel.config().setOptions(options);}
final Map<AttributeKey<?>, Object> attrs = attrs();synchronized (attrs) {for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {@SuppressWarnings("unchecked")AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();channel.attr(key).set(e.getValue());}}
ChannelPipeline p = channel.pipeline();
final EventLoopGroup currentChildGroup = childGroup;final ChannelHandler currentChildHandler = childHandler;final Entry<ChannelOption<?>, Object>[] currentChildOptions;final Entry<AttributeKey<?>, Object>[] currentChildAttrs;synchronized (childOptions) {currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));}
synchronized (childAttrs) {currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));}p.addLast(new ChannelInitializer<Channel>() {@Overridepublic void initChannel(Channel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();ChannelHandler handler = handler();if (handler != null) {pipeline.addLast(handler);}pipeline.addLast(new ServerBootstrapAcceptor(currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));}  });
}
复制代码

options() 是 serverBootstrap.option() 赋值的 AbstractBootstrap 类的 options 双向链表成员变量，在这里将 options 和 attrs 注入 channel 中 P.addLast() 为 NioServerSocketChannel 加入新的 handler (处理器)，这里 pipeline 类似于 Servlet 的过滤器，管理所有 handler

• (3.2.2) 再看 group().register() 方法
      这里的 group 是 bossGroup(NioEventLoopGroup----▷MultithreadEventLoopGroup) ,多次跳转到 SingleThreadEventLoop 类的 register() 方法

@Override
public ChannelFuture register(final Channel channel, final ChannelPromise promise) {if (channel == null) {throw new NullPointerException("channel");}if (promise == null) {throw new NullPointerException("promise");}channel.unsafe().register(this, promise);return promise;
}
复制代码

• (3.2.3) 清除一些不重要的代码，下面才是真正的注册

@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
AbstractChannel.this.eventLoop = eventLoop;if (eventLoop.inEventLoop()) {register0(promise);} else {try {eventLoop.execute(new OneTimeTask() {@Overridepublic void run() {register0(promise);}});} catch (Throwable t) {}}
}
复制代码

     eventLoop.inEventLoop() 用来判断启动线程与当前线程是否相同，相同表示已经启动，不同则有两种可能：未启动或者线程不同。

• (3.2.4) 这里线程还未启动，走 eventLoop.execute() ，这个 execute() 方法是 SingleThreadEventExecutor 类的

@Override
public void execute(Runnable task) {if (task == null) {throw new NullPointerException("task");}boolean inEventLoop = inEventLoop();if (inEventLoop) {addTask(task);} else {startThread();addTask(task);if (isShutdown() && removeTask(task)) {reject();}}if (!addTaskWakesUp && wakesUpForTask(task)) {wakeup(inEventLoop);}
}
复制代码

• (3.2.5) 启动线程

private void startThread() {if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {thread.start();}}
}
复制代码

     我们在最开始2.1里面 SingleThreadEventExecutor 构造方法内的 thread 就是在这里启动的，我们再回到2.1的

protected SingleThreadEventExecutor(EventExecutorGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp) {thread = threadFactory.newThread(new Runnable() {@Overridepublic void run() {SingleThreadEventExecutor.this.run();}}
});
复制代码

• (3.2.6) 打开 SingleThreadEventExecutor.this.run() ;

          @Overrideprotected void run() {for (;;) {boolean oldWakenUp = wakenUp.getAndSet(false);try {if (hasTasks()) {selectNow();} else {select(oldWakenUp);if (wakenUp.get()) {selector.wakeup();}}cancelledKeys = 0;needsToSelectAgain = false;final int ioRatio = this.ioRatio;if (ioRatio == 100) {processSelectedKeys();runAllTasks();} else {final long ioStartTime = System.nanoTime();processSelectedKeys();final long ioTime = System.nanoTime() - ioStartTime;runAllTasks(ioTime * (100 - ioRatio) / ioRatio);}if (isShuttingDown()) {closeAll();if (confirmShutdown()) {break;}}} catch (Throwable t) {try {Thread.sleep(1000);} catch (InterruptedException e) {}}}}
复制代码

     在这里异步执行，轮询 select 客户端的 accept ，并且 runAllTasks 所有的任务

• (3.3) 我们再看 (3.1) 里面的 ChannelFuture regFuture = group().register(channel); 跳转到 SingleThreadEventLoop 的 register 方法

@Override
public ChannelFuture register(Channel channel) {...channel.unsafe().register(this, promise);return promise;
}
复制代码

     以下是精简后的代码(位于 AbstractChannel 类的 AbstractUnsafe 内部类)

@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {...eventLoop.execute(new OneTimeTask() {@Overridepublic void run() {register0(promise);}});...
}
复制代码

private void register0(ChannelPromise promise) {...doRegister();...if (firstRegistration && isActive()) {pipeline.fireChannelActive();}...
}
复制代码

     继续(位于 AbstractNioChannel 类)

@Override
protected void doRegister() throws Exception {boolean selected = false;for (;;) {...selectionKey = javaChannel().register(eventLoop().selector, 0, this);...
}
}
复制代码

     将 NioServerSocketChannel 注册到 boss 线程池 NioEventLoop 的 Selector 上。
在这里应该注册 OP_ACCEPT(16) 到多路复用器上
注册0的原因:
     (1)注册方法是多态的，它既可以被 NioServerSocketChannel 用来监听客户端的连接接入，也可以注册 SocketChannel 用来监听网络读或写操作
    (2)通过 SelectionKey 的 interestOps(int ops) 方法可以方便地修改监听操作位

     再看 pipeline.fireChannelActive()

@Override
public ChannelPipeline fireChannelActive() {head.fireChannelActive();if (channel.config().isAutoRead()) {channel.read();}return this;
}复制代码

@Override
public Channel read() {pipeline.read();return this;
}
复制代码

@Override
public ChannelPipeline read() {tail.read();return this;
}
复制代码

@Override
public ChannelHandlerContext read() {...next.invokeRead();...
}
复制代码

private void invokeRead() {try {((ChannelOutboundHandler) handler()).read(this);} catch (Throwable t) {notifyHandlerException(t);}
}
复制代码

     进到 HeadContext 的 read

@Override
public void read(ChannelHandlerContext ctx) {unsafe.beginRead();
}
复制代码

@Override public final void beginRead() { ... doBeginRead(); ... }

@Override
protected void doBeginRead() throws Exception {
if (inputShutdown) {return;
}final SelectionKey selectionKey = this.selectionKey;
if (!selectionKey.isValid()) {return;
}
readPending = true;final int interestOps = selectionKey.interestOps();if ((interestOps & readInterestOp) == 0) {selectionKey.interestOps(interestOps | readInterestOp);}
}
复制代码

     最终在这里将 selectionKey 的监听操作位改为 OP_READ

• (4) 再看 doBind0( ) 方法

 private static void doBind0(final ChannelFuture regFuture, final Channel channel,final SocketAddress localAddress, final ChannelPromise promise) {channel.eventLoop().execute(new Runnable() {@Overridepublic void run() {if (regFuture.isSuccess()) {channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);} else {promise.setFailure(regFuture.cause());}}});
}
复制代码

     将方法丢到 reactor 线程池任务队列中执行，会先判断注册是否成功，成功则继续执行bind方法

• (5) 执行 bind( ) 方法

      @Override
public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {return pipeline.bind(localAddress, promise);
}
复制代码

@Override
public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {return tail.bind(localAddress, promise);
}
复制代码

@Override
public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {...final AbstractChannelHandlerContext next = findContextOutbound();EventExecutor executor = next.executor();...next.invokeBind(localAddress, promise);...
}
复制代码

    由于 bind 事件是出站事件，寻找出站的 handler ，执行 invokeBind( ) 方法

private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {try {((ChannelOutboundHandler) handler()).bind(this, localAddress, promise);} catch (Throwable t) {notifyOutboundHandlerException(t, promise);}
}
复制代码

@Override
public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)throws Exception {unsafe.bind(localAddress, promise);
}
复制代码

@Override
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {...doBind(localAddress);...
}
复制代码

@Override
protected void doBind(SocketAddress localAddress) throws Exception {javaChannel().socket().bind(localAddress, config.getBacklog());
}
复制代码

     经过多层 bind 深入，最后在这里可以看到，还是会调用Java底层的nio进行 socket bind 自此，服务端启动流程解析完毕，我们总结一下
     ① 通过 ServerBootstrap 辅助启动类，配置了 reactor 线程池，服务端 Channel ，一些配置参数，客户端连接后的 handler
     ② 将 ServerBootstrap 的值初始化，并注册 OP_ACCEPT 到多路复用器
     ③ 启动 reactor 线程池，不断循环监听连接，处理任务
     ④ 绑定端口