vlambda博客
学习文章列表

EventLoopGroup bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("boss", true));

public NioEventLoopGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, final SelectorProvider selectorProvider, final SelectStrategyFactory selectStrategyFactory, final RejectedExecutionHandler rejectedExecutionHandler) { super(nThreads, executor, chooserFactory, selectorProvider, selectStrategyFactory, rejectedExecutionHandler); }

protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) { if (nThreads <= 0) { throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads)); } // 上面提到的 2 if (executor == null) { executor = new ThreadPerTaskExecutor(newDefaultThreadFactory()); } // 创建 nThreads 数量的 EventExecutor 数组，通俗讲就是初始化这么多个线程 children = new EventExecutor[nThreads];
 for (int i = 0; i < nThreads; i ++) { boolean success = false; try { // 实例化 // NioEventLoopGroup#newChild 会创建 NioEventLoop // DefaultEventExecutorGroup#newChild 会创建 DefaultEventExecutor children[i] = newChild(executor, args); success = true; } catch (Exception e) { // TODO: Think about if this is a good exception type throw new IllegalStateException("failed to create a child event loop", e); } finally { // 创建失败的处理逻辑 if (!success) { for (int j = 0; j < i; j ++) { children[j].shutdownGracefully(); }
 for (int j = 0; j < i; j ++) { EventExecutor e = children[j]; try { while (!e.isTerminated()) { e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS); } } catch (InterruptedException interrupted) { // Let the caller handle the interruption. Thread.currentThread().interrupt(); break; } } } } } // 上面说的 3 chooser = chooserFactory.newChooser(children); // 给每个线程添加 termination 事件的监听器 final FutureListener<Object> terminationListener = new FutureListener<Object>() { @Override public void operationComplete(Future<Object> future) throws Exception { if (terminatedChildren.incrementAndGet() == children.length) { terminationFuture.setSuccess(null); } } };
 for (EventExecutor e: children) { e.terminationFuture().addListener(terminationListener); } // 缓存一下可读的副本，可以使用迭代器遍历 Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length); Collections.addAll(childrenSet, children); readonlyChildren = Collections.unmodifiableSet(childrenSet);}

protected EventLoop newChild(Executor executor, Object... args) throws Exception { return new NioEventLoop(this, executor, (SelectorProvider) args[0], ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);}// 也就是调用构造方法NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider, SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) { super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler); if (selectorProvider == null) { throw new NullPointerException("selectorProvider"); } if (strategy == null) { throw new NullPointerException("selectStrategy"); } provider = selectorProvider; // 每个 NioEventLoop 都会去创建一个 Selector final SelectorTuple selectorTuple = openSelector(); selector = selectorTuple.selector; unwrappedSelector = selectorTuple.unwrappedSelector; selectStrategy = strategy;}

// AbstractBootstrap#bind(SocketAddress)// AbstractBootstrap#doBind// AbstractBootstrap#initAndRegister// AbstractBootstrap#init （最终调用 ServerBootstrap#init 的实现）// 入参 NioServerSocketChannel（包装了 JDK ServerSocketChannel）void init(Channel channel) throws Exception { // 设置 Server Channel 配置和属性 final Map<ChannelOption<?>, Object> options = options0(); synchronized (options) { setChannelOptions(channel, options, logger); }
 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()); } } // 获取 Server Channel 的 pipeline ChannelPipeline p = channel.pipeline();
 // 获取 Child 的配置属性，给 channel 使用 // bootstrap.group(bossGroup, workerGroup) // .channel(NioServerSocketChannel.class) // .childOption() // .childHandler(new ChannelInitializer<NioSocketChannel>() {}); 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())); } // 往 pipeline 添加初始化 Server Channel 的 ChannelInitializer（也是一个 ChannelHandler，和其他不一样的是执行完后会 remove） p.addLast(new ChannelInitializer<Channel>() { // EventLoopGroup#register(Channel) 会调用 initChannel 方法，执行完毕后会 remove 当前 ChannelInitializer // private boolean initChannel(ChannelHandlerContext ctx) throws Exception { // if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) { // Guard against re-entrance. // try { // initChannel((C) ctx.channel()); // } catch (Throwable cause) { // // Explicitly call exceptionCaught(...) as we removed the handler before calling initChannel(...). // // We do so to prevent multiple calls to initChannel(...). // exceptionCaught(ctx, cause); // } finally { // remove(ctx); // } // return true; // } // return false; // } @Override public void initChannel(final Channel ch) throws Exception { // 真正的初始化方法 final ChannelPipeline pipeline = ch.pipeline(); ChannelHandler handler = config.handler(); if (handler != null) { pipeline.addLast(handler); }
 ch.eventLoop().execute(new Runnable() { @Override public void run() { // Acceptor，处理接受到的 SocketChannel，在下面执行流程中会说明 pipeline.addLast(new ServerBootstrapAcceptor( ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } });}

private final class NioMessageUnsafe extends AbstractNioUnsafe {
 private final List<Object> readBuf = new ArrayList<Object>();
 @Override public void read() { assert eventLoop().inEventLoop(); final ChannelConfig config = config(); final ChannelPipeline pipeline = pipeline(); final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle(); allocHandle.reset(config);
 boolean closed = false; Throwable exception = null; try { try { do { // 接受 SocketChannel int localRead = doReadMessages(readBuf); if (localRead == 0) { break; } if (localRead < 0) { closed = true; break; }
 allocHandle.incMessagesRead(localRead); } while (allocHandle.continueReading()); } catch (Throwable t) { exception = t; }
 int size = readBuf.size(); for (int i = 0; i < size; i ++) { readPending = false; // 触发读事件，主要看这个，会去调用上面提到的 ServerBootstrapAcceptor pipeline.fireChannelRead(readBuf.get(i)); } readBuf.clear(); // 其他事件，可忽略 allocHandle.readComplete(); pipeline.fireChannelReadComplete();
 if (exception != null) { closed = closeOnReadError(exception);
 pipeline.fireExceptionCaught(exception); }
 if (closed) { inputShutdown = true; if (isOpen()) { close(voidPromise()); } } } finally { if (!readPending && !config.isAutoRead()) { removeReadOp(); } } }}

public void channelRead(ChannelHandlerContext ctx, Object msg) { final Channel child = (Channel) msg; // 在上面也提到过，就是初始化 ServerBootstrap 传入的参数 // bootstrap.group(bossGroup, workerGroup) // .channel(NioServerSocketChannel.class) // .childOption() // .childHandler(new ChannelInitializer<NioSocketChannel>() {}); child.pipeline().addLast(childHandler);
 setChannelOptions(child, childOptions, logger);
 for (Entry<AttributeKey<?>, Object> e: childAttrs) { child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue()); }
 try { // childGroup 是 bootstrap.group(bossGroup, workerGroup) 传入的 workerGroup 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); }}

public ChannelFuture register(Channel channel) { // next() 获取其中一个 worker NioEventLoop，然后注册 return next().register(channel);}

ServerBootstrap childHandler = this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector) .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .option(ChannelOption.SO_REUSEADDR, true) .option(ChannelOption.SO_KEEPALIVE, false) .childOption(ChannelOption.TCP_NODELAY, true) .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize()) .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize()) .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort())) .childHandler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) throws Exception { // 这几个 addLast 方法传入了 defaultEventExecutorGroup ch.pipeline() .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, new HandshakeHandler(TlsSystemConfig.tlsMode)) .addLast(defaultEventExecutorGroup, new NettyEncoder(), new NettyDecoder(), new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()), new NettyConnectManageHandler(), new NettyServerHandler() ); } });

private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) { return new DefaultChannelHandlerContext(this, childExecutor(group), name, handler);}private EventExecutor childExecutor(EventExecutorGroup group) { if (group == null) { return null; } Boolean pinEventExecutor = channel.config().getOption(ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP); if (pinEventExecutor != null && !pinEventExecutor) { return group.next(); } Map<EventExecutorGroup, EventExecutor> childExecutors = this.childExecutors; if (childExecutors == null) { // Use size of 4 as most people only use one extra EventExecutor. childExecutors = this.childExecutors = new IdentityHashMap<EventExecutorGroup, EventExecutor>(4); } // Pin one of the child executors once and remember it so that the same child executor // is used to fire events for the same channel. // 获取一个 EventExecutor EventExecutor childExecutor = childExecutors.get(group); if (childExecutor == null) { childExecutor = group.next(); childExecutors.put(group, childExecutor); } return childExecutor;}

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) { final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next); // 获取当前 ChannelHandler 执行的 EventExecutor EventExecutor executor = next.executor(); if (executor.inEventLoop()) { next.invokeChannelRead(m); } else { // 因为触发的是 IO 线程，next.executor() 是执行 ChannelHandler 的线程，所以会走这边 executor.execute(new Runnable() { @Override public void run() { next.invokeChannelRead(m); } }); }}