netty源码分析之服务端启动

 ????????// Configure the server. ???????EventLoopGroup bossGroup = new NioEventLoopGroup(1); ???????EventLoopGroup workerGroup = new NioEventLoopGroup(); ???????final EchoServerHandler serverHandler = new EchoServerHandler(); ???????try { ???????????//初始化一个服务端引导类 ???????????ServerBootstrap b = new ServerBootstrap(); ???????????b.group(bossGroup, workerGroup) //设置线程组 ????????????.channel(NioServerSocketChannel.class)//设置ServerSocketChannel的IO模型 ?分为epoll与Nio ????????????.option(ChannelOption.SO_BACKLOG, 100)//设置option参数，保存成一个LinkedHashMap<ChannelOption<?>, Object>() ????????????.handler(new LoggingHandler(LogLevel.INFO))//这个hanlder 只专属于 ServerSocketChannel 而不是 SocketChannel。 ????????????.childHandler(new ChannelInitializer<SocketChannel>() { //这个handler 将会在每个客户端连接的时候调用。供 SocketChannel 使用。 ????????????????@Override ????????????????public void initChannel(SocketChannel ch) throws Exception { ????????????????????ChannelPipeline p = ch.pipeline(); ????????????????????if (sslCtx != null) { ????????????????????????p.addLast(sslCtx.newHandler(ch.alloc())); ????????????????????} ????????????????????//p.addLast(new LoggingHandler(LogLevel.INFO)); ????????????????????p.addLast(serverHandler); ????????????????} ????????????}); ???????????// 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(); ???????}

final ChannelFuture regFuture = initAndRegister();// 初始化并创建 NioServerSocketChannel

channel = channelFactory.newChannel(); // 通过 反射工厂创建一个 NioServerSocketChannel

 ???@Override ???public T newChannel() { ???????try { ???????????return clazz.getConstructor().newInstance();//反射创建 ???????} catch (Throwable t) { ???????????throw new ChannelException("Unable to create Channel from class " + clazz, t); ???????} ???}

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

 ???/** ????* Create a new instance ????*/ ???public NioServerSocketChannel() { ???????this(newSocket(DEFAULT_SELECTOR_PROVIDER));//传入默认的SelectorProvider ???} ???private static ServerSocketChannel newSocket(SelectorProvider provider) { ???????try { ???????????/** ????????????* ?Use the {@link SelectorProvider} to open {@link SocketChannel} and so remove condition in ????????????* ?{@link SelectorProvider#provider()} which is called by each ServerSocketChannel.open() otherwise. ????????????* ????????????* ?See <a href="https://github.com/netty/netty/issues/2308">#2308</a>. ????????????*/ ???????????return provider.openServerSocketChannel();//可以看到创建的是jdk底层的ServerSocketChannel ????????} catch (IOException e) { ???????????throw new ChannelException( ???????????????????"Failed to open a server socket.", e); ???????} ???} ???

 ???/** ????* Create a new instance using the given {@link ServerSocketChannel}. ????*/ ???public NioServerSocketChannel(ServerSocketChannel channel) { ???????super(null, channel, SelectionKey.OP_ACCEPT);//调用父类构造函数,传入创建的channel ???????config = new NioServerSocketChannelConfig(this, javaChannel().socket()); ???}

 ???/** ????* Create a new instance ????* ????* @param parent ???????????the parent {@link Channel} by which this instance was created. May be {@code null} ????* @param ch ???????????????the underlying {@link SelectableChannel} on which it operates ????* @param readInterestOp ???the ops to set to receive data from the {@link SelectableChannel} ????*/ ???protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) { ???????super(parent); ???????this.ch = ch;//这个ch就是传入的通过jdk创建的Channel ???????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); ???????????????} ???????????} ???????????throw new ChannelException("Failed to enter non-blocking mode.", e); ???????} ???}

 ???/** ????* Creates a new instance. ????* ????* @param parent ????* ???????the parent of this channel. {@code null} if there‘s no parent. ????*/ ???protected AbstractChannel(Channel parent) { ???????this.parent = parent; ???????id = newId();//创建Channel唯一标识 ????????unsafe = newUnsafe();//netty封装的TCP 相关操作类 ???????pipeline = newChannelPipeline();//逻辑链 ???}

init(channel);// 初始化这个 NioServerSocketChannel

 ???????final Map<ChannelOption<?>, Object> options = options0();//拿到你设置的option ???????synchronized (options) { ???????????setChannelOptions(channel, options, logger);//设置NioServerSocketChannel相应的TCP参数，其实这一步就是把options设置到channel的config中 ???????} ???????final Map<AttributeKey<?>, Object> attrs = attrs0(); ???????synchronized (attrs) { ???????????for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) { ???????????????@SuppressWarnings("unchecked") ???????????????AttributeKey<Object> key = (AttributeKey<Object>) e.getKey(); ???????????????channel.attr(key).set(e.getValue()); ???????????} ???????}

 ???????//可以看到两个都是局部变量，会在下面设置pipeline时用到 ???????final Entry<ChannelOption<?>, Object>[] currentChildOptions; ???????final Entry<AttributeKey<?>, Object>[] currentChildAttrs; ???????synchronized (childOptions) { ???????????currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0)); ???????} ???????synchronized (childAttrs) { ???????????currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0)); ???????}

 ???????p.addLast(new ChannelInitializer<Channel>() { ???????????@Override ???????????public void initChannel(final Channel ch) throws Exception { ???????????????final ChannelPipeline pipeline = ch.pipeline(); ???????????????ChannelHandler handler = config.handler();//拿到我们自定义的hanler ???????????????if (handler != null) { ???????????????????pipeline.addLast(handler); ???????????????} ???????????????ch.eventLoop().execute(new Runnable() { ???????????????????@Override ???????????????????public void run() { ???????????????????????pipeline.addLast(new ServerBootstrapAcceptor( ???????????????????????????????ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); ???????????????????} ???????????????}); ???????????} ???????});

 ???????????????ch.eventLoop().execute(new Runnable() { ???????????????????@Override ???????????????????public void run() { ???????????????????????//在这里会把我们自定义的ChildGroup、ChildHandler、ChildOptions、ChildAttrs相关配置传入到ServerBootstrapAcceptor构造函数中，并绑定到新的连接上 ???????????????????????pipeline.addLast(new ServerBootstrapAcceptor( ???????????????????????????????ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); ???????????????????} ???????????????});

ChannelFuture regFuture = config().group().register(channel);//注册我们已经初始化过的Channel

 /** ????????* 1、先是一系列的判断。 ????????* 2、判断当前线程是否是给定的 eventLoop 线程。注意：这点很重要，Netty 线程模型的高性能取决于对于当前执行的Thread 的身份的确定。如果不在当前线程，那么就需要很多同步措施（比如加锁），上下文切换等耗费性能的操作。 ????????* 3、异步（因为我们这里直到现在还是 main 线程在执行，不属于当前线程）的执行 register0 方法。 ????????*/ ???????@Override ???????public final void register(EventLoop eventLoop, final ChannelPromise promise) { ???????????if (eventLoop == null) { ???????????????throw new NullPointerException("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; ???????????} ???????????AbstractChannel.this.eventLoop = 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); ???????????????} ???????????} ???????}

 ???????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 channel的底层注册 ???????????????neverRegistered = false; ???????????????registered = true; ???????????????// 触发绑定的handler事件 ???????????????// 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(); ???????????????// 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 ???protected void doRegister() throws Exception { ???????boolean selected = false; ???????for (;;) { ???????????try { ???????????????//把channel注册到eventLoop上的selector上 ???????????????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; ???????????????} ???????????} ???????} ???}

 @Override ???????public final void bind(final SocketAddress localAddress, final ChannelPromise promise) { ???????????assertEventLoop(); ???????????if (!promise.setUncancellable() || !ensureOpen(promise)) { ???????????????return; ???????????} ???????????// See: https://github.com/netty/netty/issues/576 ???????????if (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);//底层jdk绑定端口 ???????????} catch (Throwable t) { ???????????????safeSetFailure(promise, t); ???????????????closeIfClosed(); ???????????????return; ???????????} ???????????if (!wasActive && isActive()) { ???????????????invokeLater(new Runnable() { ???????????????????@Override ???????????????????public void run() { ???????????????????????pipeline.fireChannelActive();//触发ChannelActive事件 ???????????????????} ???????????????}); ???????????} ???????????safeSetSuccess(promise); ???????}

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

 ???????@Override ???????public void channelActive(ChannelHandlerContext ctx) throws Exception { ???????????ctx.fireChannelActive(); ???????????readIfIsAutoRead(); ???????}

 ???@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(); ???????if ((interestOps & readInterestOp) == 0) { ???????????selectionKey.interestOps(interestOps | readInterestOp);//readInterestOp为 ?SelectionKey.OP_ACCEPT ???????} ???}