vlambda博客
学习文章列表

serverBootstrap.bind(port).addListener(future -> { if(future.isSuccess()) { System.out.println(newDate() + ": 端口["+ port + "]绑定成功!"); } else{ System.err.println("端口["+ port + "]绑定失败!"); } });

public B channel(Class<? extends C> channelClass) { /** * 这里将传入的比如NioServerSocketChannel通过反射创建一个ReflectiveChannelFactory对象，然后 * 调用channelFactory将创建的ReflectiveChannelFactory对象赋值给ServerBootstrap 中channelFactory属性 * 所以这里记录下，在ServerBootstrap中有个属性为channelFactory * 这里的new ReflectiveChannelFactory其实就是得到了传入进来的channelClass一个对象，通过反射调用 * 构造NioServerSocketChannel的构造空构造方法 */ return channelFactory(new ReflectiveChannelFactory<C>( ObjectUtil.checkNotNull(channelClass, "channelClass") ));}
io.netty.channel.socket.nio.NioServerSocketChannel#NioServerSocketChannel()/** * Create a new instance * 这里是一个空构造方法，在这个构造方法中做的事情有： * 1.创建一个ServerSocketChannel（NIO）； * 2.将创建的ServerSocketChannel设置一些参数，比如设置这个ServerSocketChannel的事件为OP_ACCEPT，就是接受连接事件 * 3.设置ServerSocketChannel为非阻塞的模式。 */public NioServerSocketChannel() { //newSocket就是创建一个ServerSocketChannel，调用的就是NIO的创建逻辑 this(newSocket(DEFAULT_SELECTOR_PROVIDER));}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>. * 调用NIO创建一个ServerSocketChannel */ return provider.openServerSocketChannel(); } catch (IOException e) { throw new ChannelException( "Failed to open a server socket.", e); }}public NioServerSocketChannel(ServerSocketChannel channel) { //调用父类将这个ServerSocketChannel设置为非阻塞模式和将创建的ServerSocketChannel赋值到父类的属性ch中 super(null, channel, SelectionKey.OP_ACCEPT); config = new NioServerSocketChannelConfig(this, javaChannel().socket());}//AbstractNioChannel是NioServerSocketChannel的父类protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) { //调用父类初始化的管道ChannelPipeline super(parent); //赋值channel this.ch = ch; //设置这个ServerSocketChannel的事件类型为readInterestOp（OP_ACCEPT) this.readInterestOp = readInterestOp; try { //设置这个ServerSocketChannel为非阻塞的模式 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); }}//AbstractChannel是AbstractNioChannel的父类protected AbstractChannel(Channel parent) { this.parent = parent; //创建一个ChannelId id = newId(); //创建AbstractUnsafe对象，后续要使用 unsafe = newUnsafe(); //初始化管道pipeline对象，这个对象非常重要，就是我们的管道的处理器就是放在里面的，所有的处理器都会加入到这个管道中 //这个管道就是一个双向链表，有head和tail,默认的管道实现类是DefaultChannelPipeline pipeline = newChannelPipeline();}
io.netty.channel.DefaultChannelPipeline#DefaultChannelPipeline//channel就是刚刚NioServerSocketChannel，也是一个ServerScoketChannelprotected DefaultChannelPipeline(Channel channel) { this.channel = ObjectUtil.checkNotNull(channel, "channel"); succeededFuture = new SucceededChannelFuture(channel, null); voidPromise = new VoidChannelPromise(channel, true);
 //这个管道pipeline中的尾部 tail = new TailContext(this); //这个管道pipeline中的头部 head = new HeadContext(this);
 //建立双向链表关系 head.next = tail; tail.prev = head;}

/** * 这个方法就是Netty的启动过程的最核心的方法，前面的都只是为 下面这个方法做准备的 * 这个方法里面有两个最核心的方法， * initAndRegister：初始化和注册,注册的是将创建ServerSocketChannel注册到多路复用器selector上 * doBind0：端口绑定，启动监听 * 总结来说就是initAndRegister配置Netty和注册多路复用器 * doBind0：绑定端口，调用epoll阻塞,监听事件的发生 * @param localAddress * @return */private ChannelFuture doBind(final SocketAddress localAddress) {
 //初始化和注册（太多 ） final ChannelFuture regFuture = initAndRegister(); final Channel channel = regFuture.channel(); if (regFuture.cause() != null) { return regFuture; }
 /** * 下面的代码逻辑就是绑定端口，开启select阻塞 * 首先条件isDone第一次启动是不会进入这里，走的else的逻辑 * 在else逻辑中添加了一个监听器，最后由监听器出发了doBind0方法 * 所以不管怎么说，核心代码逻辑都是doBind0 */ 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); //添加一个监听器，最后由监听器调用了doBind0方法 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();
 /** * 在DefaultChannelPromise中的notifyListeners进行调用的 */
 doBind0(regFuture, channel, localAddress, promise); } } }); return promise; }}

/** * 多路复用器注册的核心方法 * @param task */@Overridepublic void execute(Runnable task) { ObjectUtil.checkNotNull(task, "task"); execute(task, !(task instanceof LazyRunnable) && wakesUpForTask(task));}/** * 多路复用器注册的核心方法，task就是最外层的那个register0所在的线程 * @param task * @param immediate */private void execute(Runnable task, boolean immediate) { //这里就是和外层的那个判断一样，就是看当前线程是否是NioEventLoop的那个线程，代码到这里还在主线程中，所以这里返回的false //因为task线程还没开始执行 boolean inEventLoop = inEventLoop(); //将注册的那个register0所在线程加入到task任务中,也就是加入到taskQueue.offer(task)队列中，等待调度 addTask(task); if (!inEventLoop) { //开始线程 startThread(); if (isShutdown()) { boolean reject = false; try { if (removeTask(task)) { reject = true; } } catch (UnsupportedOperationException e) { // The task queue does not support removal so the best thing we can do is to just move on and // hope we will be able to pick-up the task before its completely terminated. // In worst case we will log on termination. } if (reject) { reject(); } } }
 if (!addTaskWakesUp && immediate) { wakeup(inEventLoop); }}private void startThread() { if (state == ST_NOT_STARTED) { if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) { boolean success = false; try { doStartThread(); success = true; } finally { if (!success) { STATE_UPDATER.compareAndSet(this, ST_STARTED, ST_NOT_STARTED); } } } }}

@Overridepublic 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)) { promise.setFailure( new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName())); return; }
 //eventLoop 是从boss 线程组选择出来一个NioEventLoop AbstractChannel.this.eventLoop = eventLoop;
 //下面这个if表示当前的线程是否和NioEventLoop是否在一个线程中的 //反正就是判断需要同步处理还是异步处理，Netty是一个异步非阻塞的通信框架，所以很多地方都是使用这种线程池的方案去处理的 //所以毫无疑问，register0是核心方法 if (eventLoop.inEventLoop()) { register0(promise); } else { try { /** * 这里不要被这个execute给迷糊了，看到这个以为是一个线程的启动方法，其实不是，其实就是简单的 * 调用了对象eventLoop中的execute方法，这个方法传入了一个线程对象，但是肯定不是在这里执行的这个线程的run方法 * 的，所以这里记住这个execute方法中传入了一个内部类线程，肯定要先看这个execute方法，而这里的 * eventLoop是NioEventLoop,也是SingleThreadEventExecutor */ 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); } }}
io.netty.util.concurrent.SingleThreadEventExecutor#execute(java.lang.Runnable)/** * 多路复用器注册的核心方法 * @param task */@Overridepublic void execute(Runnable task) { ObjectUtil.checkNotNull(task, "task"); execute(task, !(task instanceof LazyRunnable) && wakesUpForTask(task));}/** * 多路复用器注册的核心方法，task就是最外层的那个register0所在的线程 * @param task * @param immediate */private void execute(Runnable task, boolean immediate) { //这里就是和外层的那个判断一样，就是看当前线程是否是NioEventLoop的那个线程，代码到这里还在主线程中，所以这里返回的false //因为task线程还没开始执行 boolean inEventLoop = inEventLoop(); //将注册的那个register0所在线程加入到task任务中,也就是加入到taskQueue.offer(task)队列中，等待调度 addTask(task); if (!inEventLoop) { //开始线程 startThread(); if (isShutdown()) { boolean reject = false; try { if (removeTask(task)) { reject = true; } } catch (UnsupportedOperationException e) { // The task queue does not support removal so the best thing we can do is to just move on and // hope we will be able to pick-up the task before its completely terminated. // In worst case we will log on termination. } if (reject) { reject(); } } }
 if (!addTaskWakesUp && immediate) { wakeup(inEventLoop); }}private void startThread() { if (state == ST_NOT_STARTED) { if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) { boolean success = false; try { doStartThread(); success = true; } finally { if (!success) { STATE_UPDATER.compareAndSet(this, ST_STARTED, ST_NOT_STARTED); } } } }}

/** * 这里的run方法才是真正的将NIOServerSocketChannel注册到多路复用器上 */@Overrideprotected void run() { int selectCnt = 0; for (;;) { try { int strategy; try { /** *这里就是判断现在是那种事件，如果Netty在启动的时候，这里的case都不满足，但是如果启动完成了 *那么再来调用就会调用SELECT，所以根据不同的情况来的 */ strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks()); switch (strategy) { case SelectStrategy.CONTINUE: continue;
 case SelectStrategy.BUSY_WAIT: // fall-through to SELECT since the busy-wait is not supported with NIO
 case SelectStrategy.SELECT: //这里就是调用的多路复用器的select方法，就是在nio中的selector.select()方法 /** * curDeadlineNanos这个是计算调用select的时候阻塞的时间的，因为netty这里 还有很多事情没有做 * 注册多路复用器都还没有做，所以这里要计算出一个时间，如果这个时间过了，还没有连接过来，那么也会唤醒, * 不会一直阻塞 */ long curDeadlineNanos = nextScheduledTaskDeadlineNanos(); if (curDeadlineNanos == -1L) { curDeadlineNanos = NONE; // nothing on the calendar } nextWakeupNanos.set(curDeadlineNanos); try { /** * 如果是第一次启动的注册多路复用器是不会进入这个 if的，因为 * hasTasks就是判断当前的多路复用器注册队列中是是否有任务，如果有任务， * 那么返回true，没有就返回 false的，netty的意思就是服务端刚启动要注册的是NioServerSocketChannel * 所以不阻塞，否则阻塞了没意义，因为多路复用器都还没有注册的有channel，根本没有办法监听 */ if (!hasTasks()) { strategy = select(curDeadlineNanos); } } finally { // This update is just to help block unnecessary selector wakeups // so use of lazySet is ok (no race condition) nextWakeupNanos.lazySet(AWAKE); } // fall through default: } } catch (IOException e) { // If we receive an IOException here its because the Selector is messed up. Let's rebuild // the selector and retry. https://github.com/netty/netty/issues/8566 rebuildSelector0(); selectCnt = 0; handleLoopException(e); continue; }
 /** * strategy在Netty启动的时候是0 */ selectCnt++; cancelledKeys = 0; needsToSelectAgain = false; final int ioRatio = this.ioRatio; boolean ranTasks; if (ioRatio == 100) { try { if (strategy > 0) { //调用select有事件发生的时候会来调用，这里面就是NIO的那个获取selectKeys然后处理的逻辑，netty封装了 processSelectedKeys(); } } finally { // Ensure we always run tasks. ranTasks = runAllTasks(); } } else if (strategy > 0) { final long ioStartTime = System.nanoTime(); try { //调用select有事件发生的时候会来调用 processSelectedKeys(); } finally { // Ensure we always run tasks. final long ioTime = System.nanoTime() - ioStartTime; ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio); } } else { //Netty启动的时候调用，这里才是真正的多路复用器的注册，就是调用最开始的那个execute（Runable task）中的那个task //就是从taskQueue中取出这个线程任务然后执行run方法 ranTasks = runAllTasks(0); // This will run the minimum number of tasks }
 if (ranTasks || strategy > 0) { if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) { logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.", selectCnt - 1, selector); } selectCnt = 0; } else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case) selectCnt = 0; } } catch (CancelledKeyException e) { // Harmless exception - log anyway if (logger.isDebugEnabled()) { logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?", selector, e); } } catch (Error e) { throw (Error) e; } catch (Throwable t) { handleLoopException(t); } finally { // Always handle shutdown even if the loop processing threw an exception. try { if (isShuttingDown()) { closeAll(); if (confirmShutdown()) { return; } } } catch (Error e) { throw (Error) e; } catch (Throwable t) { handleLoopException(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; //这里就是将NioServerSocketChannel注册到多路复用上Selector 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. /** * 这个方法的调用就是处理管道的handler的，就是处理器， * 它的过程会将之前创建的ChannelInitializer那个管道中的处理器添加到管道中，然后删除 * 简单来说就是先将ChannelInitializer中的那个initChannel方法调用完成过后，然后删除这个ChannelInitializer那 * 内部类 */ pipeline.invokeHandlerAddedIfNeeded();
 safeSetSuccess(promise); //这里的调用的注册，那么注册也要调用管道中的所有的有实现了注册的处理器，调用是从链表的head开始的 //当看到有调用fireChannelRegistered方法的， 那么证明是调用了下一个处理器的ChannelRegistered方法 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()) { //表示netty现在是激活的状态来调用的 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); }}@Overrideprotected void doRegister() throws Exception { boolean selected = false; for (;;) { try { //将NioServerSocketChannel 注册到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; } } }}