Netty Nio启动全流程
1. 各组件之间的关系
说明：EventLoopGroup类似线程池，EventLoop为单线程，每个EventLoop关联一个Nio Selector,用于注册Channel，形成一个EventLoop被多个channel公用。在EventLoop会执行通道Io选择操作，以及非Io任务。在Channel初始化后会创建pipeline,是handler的链表结构。
2. 服务端vs客户端启动
1// 服务端启动 2private ChannelFuture doBind(final SocketAddress localAddress) { 3 final ChannelFuture regFuture = initAndRegister(); 4 final Channel channel = regFuture.channel(); 5 6 if (regFuture.cause() != null) { 7 return regFuture; 8 } 9 10 if (regFuture.isDone()) { 11 // At this point we know that the registration was complete and successful. 12 ChannelPromise promise = channel.newPromise(); 13 doBind0(regFuture, channel, localAddress, promise); 14 return promise; 15 } else { 16 // Registration future is almost always fulfilled already, but just in case it's not. 17 final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel); 18 regFuture.addListener(new ChannelFutureListener() { 19 @Override 20 public void operationComplete(ChannelFuture future) throws Exception { 21 Throwable cause = future.cause(); 22 if (cause != null) { 23 // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an 24 // IllegalStateException once we try to access the EventLoop of the Channel. 25 promise.setFailure(cause); 26 } else { 27 // Registration was successful, so set the correct executor to use. 28 // See https://github.com/netty/netty/issues/2586 29 promise.registered(); 30 31 doBind0(regFuture, channel, localAddress, promise); 32 } 33 } 34 }); 35 return promise; 36 } 37} 38 39 40//客户端启动 41private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) { 42 final ChannelFuture regFuture = initAndRegister(); 43 final Channel channel = regFuture.channel(); 44 45 if (regFuture.isDone()) { 46 if (!regFuture.isSuccess()) { 47 return regFuture; 48 } 49 return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise()); 50 } else { 51 // Registration future is almost always fulfilled already, but just in case it's not. 52 final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel); 53 regFuture.addListener(new ChannelFutureListener() { 54 @Override 55 public void operationComplete(ChannelFuture future) throws Exception { 56 // Directly obtain the cause and do a null check so we only need one volatile read in case of a 57 // failure. 58 Throwable cause = future.cause(); 59 if (cause != null) { 60 // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an 61 // IllegalStateException once we try to access the EventLoop of the Channel. 62 promise.setFailure(cause); 63 } else { 64 // Registration was successful, so set the correct executor to use. 65 // See https://github.com/netty/netty/issues/2586 66 promise.registered(); 67 doResolveAndConnect0(channel, remoteAddress, localAddress, promise); 68 } 69 } 70 }); 71 return promise; 72 } 73}
一言以蔽之，首先做初始化channel和channel注册操作，然后服务器启动做绑定操作，客户端启动做连接操作。而初始化channel和channel注册都是通过initAndRegister()实现。最大化重用代码。
3. 初始化创建通道以及通道注册
3.1 模板方法的创建通道->初始化通道->通道注册
1final ChannelFuture initAndRegister() { 2 Channel channel = null; 3 try { 4 // 创建通道 5 channel = channelFactory.newChannel(); 6 // 初始化通道 7 init(channel); 8 } catch (Throwable t) { 9 if (channel != null) { 10 // channel can be null if newChannel crashed (eg SocketException("too many open files")) 11 channel.unsafe().closeForcibly(); 12 // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor 13 return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t); 14 } 15 // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor 16 return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t); 17 } 18 19 // 通道注册 20 ChannelFuture regFuture = config().group().register(channel); 21 if (regFuture.cause() != null) { 22 if (channel.isRegistered()) { 23 channel.close(); 24 } else { 25 channel.unsafe().closeForcibly(); 26 } 27 } 28 return regFuture; 29}
3.2 创建通道
-
构造channel
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) { super(parent); this.ch = ch; this.readInterestOp = readInterestOp; try { ch.configureBlocking(false); } catch (IOException e) { try { ch.close(); } catch (IOException e2) { if (logger.isWarnEnabled()) { logger.warn( "Failed to close a partially initialized socket.", e2); } }
1 throw new ChannelException("Failed to enter non-blocking mode.", e); 2}}
protected AbstractChannel(Channel parent) { this.parent = parent; id = newId(); unsafe = newUnsafe(); pipeline = newChannelPipeline(); }
NioChannel将java SelectableChannel包装了一把，并添加了pipeline和unsafe操作，默认的pipeline是一个双向链表结构，只包含head和tail两个节点。 2. 初始化channel
对于客户端而言，直接向pipeline中添加builder方法的handler,以及一些nio操作的通用属性，对于服务端创建而言，除了一些基本nio属性外，只添加了一个初始化的handler
1// 客户端创建 2ChannelPipeline p = channel.pipeline(); 3p.addLast(config.handler()); 4 5 6//服务端创建 7p.addLast(new ChannelInitializer<Channel>() { 8 @Override 9 public void initChannel(final Channel ch) throws Exception { 10 final ChannelPipeline pipeline = ch.pipeline(); 11 ChannelHandler handler = config.handler(); 12 if (handler != null) { 13 pipeline.addLast(handler); 14 } 15 16 ch.eventLoop().execute(new Runnable() { 17 @Override 18 public void run() { 19 pipeline.addLast(new ServerBootstrapAcceptor( 20 ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); 21 } 22 }); 23 } 24});
注：ChannelInitializer的initChannnel会在注册成功之后调用，以此实现动态扩展。 客户端创建时候pipeline中没有ChannelInitializer，需要自己添加。 3. 通道注册
主要将channel绑定到EventLoop上面，然后在eventLoop单线程中执行注册操作
1@Override 2public final void register(EventLoop eventLoop, final ChannelPromise promise) { 3 if (eventLoop == null) { 4 throw new NullPointerException("eventLoop"); 5 } 6 if (isRegistered()) { 7 promise.setFailure(new IllegalStateException("registered to an event loop already")); 8 return; 9 } 10 if (!isCompatible(eventLoop)) { 11 promise.setFailure( 12 new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName())); 13 return; 14 } 15 16 AbstractChannel.this.eventLoop = eventLoop; 17 18 // 此时在主线程中，不知eventLoop线程池中 19 if (eventLoop.inEventLoop()) { 20 register0(promise); 21 } else { 22 try { 23 eventLoop.execute(new Runnable() { 24 @Override 25 public void run() { 26 register0(promise); 27 } 28 }); 29 } catch (Throwable t) { 30 logger.warn( 31 "Force-closing a channel whose registration task was not accepted by an event loop: {}", 32 AbstractChannel.this, t); 33 closeForcibly(); 34 closeFuture.setClosed(); 35 safeSetFailure(promise, t); 36 } 37 } 38} 39
register0主要干三件事，注册->调用ChannelInitializer的initChannnel完成添加handler->注册channel关心的操作 3.1 java channel注册,0表示只注册，不执行任何操作
selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
3.2 pipeline.fireChannelRegistered() 此时，pipeline中包含三个handler，其中一个是ChannelInitializer。
1public final void channelRegistered(ChannelHandlerContext ctx) throws Exception { 2 if (initChannel(ctx)) { 3 ctx.pipeline().fireChannelRegistered(); 4 } else { 5 ctx.fireChannelRegistered(); 6 } 7}
3.2 beginRead();
1@Override 2protected void doBeginRead() throws Exception { 3 // Channel.read() or ChannelHandlerContext.read() was called 4 final SelectionKey selectionKey = this.selectionKey; 5 if (!selectionKey.isValid()) { 6 return; 7 } 8 9 readPending = true; 10 11 final int interestOps = selectionKey.interestOps(); 12 if ((interestOps & readInterestOp) == 0) { 13 selectionKey.interestOps(interestOps | readInterestOp); 14 } 15}
注意，此时才会真实注册关系的事件，对服务端而言为Accept,对客户端创建，就是connect
1public NioServerSocketChannel(ServerSocketChannel channel) { 2 super(null, channel, SelectionKey.OP_ACCEPT); 3 config = new NioServerSocketChannelConfig(this, javaChannel().socket()); 4 } 5 6 7protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) { 8 super(parent, ch, SelectionKey.OP_READ); 9 }
至此，客户端与服务端完成了初始化channel以及注册channel操作。
4. 服务端绑定到指定端口
1private static void doBind0( 2 final ChannelFuture regFuture, final Channel channel, 3 final SocketAddress localAddress, final ChannelPromise promise) { 4 channel.eventLoop().execute(new Runnable() { 5 @Override 6 public void run() { 7 if (regFuture.isSuccess()) { 8 channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE); 9 } else { 10 promise.setFailure(regFuture.cause()); 11 } 12 } 13 }); 14}
在eventLoop中执行绑定端口操作
1@Override 2public void bind( 3 ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) 4 throws Exception { 5 unsafe.bind(localAddress, promise); 6}
最后都是会调用unsafe的bind方法完成端口绑定操作。
5. 客户端连接远程服务端
1private static void doConnect( 2 final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise connectPromise) { 3 final Channel channel = connectPromise.channel(); 4 channel.eventLoop().execute(new Runnable() { 5 @Override 6 public void run() { 7 if (localAddress == null) { 8 channel.connect(remoteAddress, connectPromise); 9 } else { 10 channel.connect(remoteAddress, localAddress, connectPromise); 11 } 12 connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE); 13 } 14 }); 15}
连接服务端最终也是在eventLoop中执行，最终调用unsafe的connect方法。
1protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception { 2 if (localAddress != null) { 3 doBind0(localAddress); 4 } 5 6 boolean success = false; 7 try { 8 boolean connected = SocketUtils.connect(javaChannel(), remoteAddress); 9 if (!connected) { 10 selectionKey().interestOps(SelectionKey.OP_CONNECT); 11 } 12 success = true; 13 return connected; 14 } finally { 15 if (!success) { 16 doClose(); 17 } 18 } 19}
connect有三种结果，成功，直接返回true,失败则暂时不知道结果，检测OP_CONNECT,异常直接关闭链路。
值得说明的是jdk默认不支持连接超时，netty添加了超时机制：在EventLoop中添加超时任务，触发超时时间后会关闭连接，连接成功会删除该超时任务。
1// Schedule connect timeout. 2int connectTimeoutMillis = config().getConnectTimeoutMillis(); 3if (connectTimeoutMillis > 0) { 4 connectTimeoutFuture = eventLoop().schedule(new Runnable() { 5 @Override 6 public void run() { 7 ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise; 8 ConnectTimeoutException cause = 9 new ConnectTimeoutException("connection timed out: " + remoteAddress); 10 if (connectPromise != null && connectPromise.tryFailure(cause)) { 11 close(voidPromise()); 12 } 13 } 14 }, connectTimeoutMillis, TimeUnit.MILLISECONDS); 15} 16 17promise.addListener(new ChannelFutureListener() { 18 @Override 19 public void operationComplete(ChannelFuture future) throws Exception { 20 if (future.isCancelled()) { 21 if (connectTimeoutFuture != null) { 22 connectTimeoutFuture.cancel(false); 23 } 24 connectPromise = null; 25 close(voidPromise()); 26 } 27 } 28}); 29
6.EventLooop 处理IO事件
1if ((readyOps & SelectionKey.OP_CONNECT) != 0) { 2 // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking 3 // See https://github.com/netty/netty/issues/924 4 int ops = k.interestOps(); 5 ops &= ~SelectionKey.OP_CONNECT; 6 k.interestOps(ops); 7 8 unsafe.finishConnect(); 9} 10 11// Process OP_WRITE first as we may be able to write some queued buffers and so free memory. 12if ((readyOps & SelectionKey.OP_WRITE) != 0) { 13 // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write 14 ch.unsafe().forceFlush(); 15} 16 17// Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead 18// to a spin loop 19if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) { 20 unsafe.read(); 21}