Netty4源码初学习-服务端启动

本文基于netty-all:4.1.6.Final版本进行简略分析，如有不对，感谢指出。
初次接触netty，网上发现一张比较好的线程模型，如图1

图1

从图1中可以看出，抛开细节，整个netty的运转是类似于java线程池，不断地执行任务队列中的任务，后面源码中会分析，线程池的继承关系。

下面就按照图1进行一步步分析，我们服务端常见的启动代码如下：

scss复制代码 public void start(InetSocketAddress address){
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup(5);
        try {
            ServerBootstrap bootstrap = new ServerBootstrap()
                    .group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .localAddress(address)
//                    .handler(null)
                    .handler(new TestHander())
                    .childHandler(new ServerPojoChannelInitializer())
//                    .childHandler(new ServerChannelInitializer())
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
            // 绑定端口，开始接收进来的连接
            ChannelFuture future = bootstrap.bind(address).sync();
            System.out.println("Server start listen at " + address.getPort());
            future.channel().closeFuture().sync();
            System.out.println("******");
        } catch (Exception e) {
            e.printStackTrace();
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

1. BossGroup初始化工作

NioEventLoopGroup的继承关系如下

可以看出group其实最顶层服务类Exector,那么其必然是通过线程对各种任务进行管理。

在其父类MultithreadEventLoopGroup构造函数中，如果不传值，就采用默认2cpu数量

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

在向上MultithreadEventExecutorGroup类中，其构造函数如下：

ini复制代码protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                        EventExecutorChooserFactory chooserFactory, Object... args) {
    if (nThreads <= 0) {
        throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
    }

    if (executor == null) {
    //通过工厂类DefaultThreadFactory生成FastThreadLocalThread线程类
        executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
    }

    children = new EventExecutor[nThreads];

    for (int i = 0; i < nThreads; i ++) {
        boolean success = false;
        try {
        //每一个channel都会有有一个线程池与之对应，这个nThereads就是初始化时传入的参数
            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;
                    }
                }
            }
        }
    }

    //线程选择策略，有两种，PowerOfTowEventExecutorChooser和GenericEventExecutorChooser
    chooser = chooserFactory.newChooser(children);

    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);
}

NioEventLoop的参数就可以看到，图1中的Selector和TaskQueue

到这里基本就看到taskQueue和selector如何传入。

2. 服务端channel初始化和注入到eventloop

由图1可知道，channel和某一个eventloop是一一对应的关系，服务端初始化好了，那下面就看看channel是如何绑定到eventloop的，直接从bind接口入手，找到AbstractBootstrap的initAndRegister方法。

scss复制代码final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
    //工厂方法获取的是启动代码中赋值的NioServerSocketChannel
        channel = channelFactory.newChannel();
        init(channel);
    } catch (Throwable t) {
        if (channel != null) {
            // channel can be null if newChannel crashed (eg SocketException("too many open files"))
            channel.unsafe().closeForcibly();
        }
        // as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
        return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
    }

    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }

    // If we are here and the promise is not failed, it's one of the following cases:
    // 1) If we attempted registration from the event loop, the registration has been completed at this point.
    //    i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
    // 2) If we attempted registration from the other thread, the registration request has been successfully
    //    added to the event loop's task queue for later execution.
    //    i.e. It's safe to attempt bind() or connect() now:
    //         because bind() or connect() will be executed *after* the scheduled registration task is executed
    //         because register(), bind(), and connect() are all bound to the same thread.

    return regFuture;
}

2.1 init方法

看下初始化方法做了啥，

scss复制代码@Override
void init(Channel channel) throws Exception {
   final Map<ChannelOption<?>, Object> options = options0();
   synchronized (options) {
       channel.config().setOptions(options);
   }

   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());
       }
   }
   //每个channel都会绑定一个pipeline
   ChannelPipeline p = channel.pipeline();

//客户端eventloop,对应图1的workgroup
   final EventLoopGroup currentChildGroup = childGroup;
   //workgroup对应的handler,上行和下行，也即启动代码中的ServerPojoChannelInitializer
   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()));
   }

//添加一个channelHandler，待后面启动
   p.addLast(new ChannelInitializer<Channel>() {
       @Override
       public void initChannel(Channel ch) throws Exception {
           final ChannelPipeline pipeline = ch.pipeline();
           ChannelHandler handler = config.handler();
           if (handler != null) {
               pipeline.addLast(handler);
           }

           // We add this handler via the EventLoop as the user may have used a ChannelInitializer as handler.
           // In this case the initChannel(...) method will only be called after this method returns. Because
           // of this we need to ensure we add our handler in a delayed fashion so all the users handler are
           // placed in front of the ServerBootstrapAcceptor.
               //前面说到，此处是netty自己实现的execute,如果执行线程和eventloop不是同一个线程那么就启动eventloop线程，否则就将任务添加到任务队列，也即taskQueue
           ch.eventLoop().execute(new Runnable() {
               @Override
               public void run() {
               //这个很关键，任务启动后，就会不断监听事件，然后再对对应的处理，也就是图1中event group下部的圆形循环
                   pipeline.addLast(new ServerBootstrapAcceptor(
                           currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
               }
           });
       }
   });
}

从上面可以，init方法，就是将读写的event loop加载进来，待触发任务回调ServerBootstrapAcceptor的read方法，不断轮询监听scoket事件

2.1 register方法

init方法并没有触发监听任务，由此可以大体猜测到
将服务端channel注册到event loop即可实现监听。由bootstrap初始化可知，其eventloop为NioEventLoop，AbstractChannel的register方法

arduino复制代码@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;
    //启动时，是main方法执行此函数
    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);
        }
    }
}

首次启动，会执行eventLoop的execute方法，看下实现SingleThreadEventExecutor.execute

scss复制代码@Override
public void execute(Runnable task) {
    if (task == null) {
        throw new NullPointerException("task");
    }

    boolean inEventLoop = inEventLoop();
    if (inEventLoop) {
        addTask(task);
    } else {
    //对于服务端启动来说，最开始是main函数执行
    //启动后，会将event的线程置为执行他的线程，也就是boss group初始化new出来的线程
        startThread();
        addTask(task);
        if (isShutdown() && removeTask(task)) {
            reject();
        }
    }

    if (!addTaskWakesUp && wakesUpForTask(task)) {
        wakeup(inEventLoop);
    }
}

startThread是采用cas来保证并发安全的

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

然后能看到nioeventloop的run方法是个死循环

scss复制代码@Override
protected void run() {
    for (;;) {
        try {
            switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                case SelectStrategy.CONTINUE:
                    continue;
                case SelectStrategy.SELECT:
                    select(wakenUp.getAndSet(false));

                    // 'wakenUp.compareAndSet(false, true)' is always evaluated
                    // before calling 'selector.wakeup()' to reduce the wake-up
                    // overhead. (Selector.wakeup() is an expensive operation.)
                    //
                    // However, there is a race condition in this approach.
                    // The race condition is triggered when 'wakenUp' is set to
                    // true too early.
                    //
                    // 'wakenUp' is set to true too early if:
                    // 1) Selector is waken up between 'wakenUp.set(false)' and
                    //    'selector.select(...)'. (BAD)
                    // 2) Selector is waken up between 'selector.select(...)' and
                    //    'if (wakenUp.get()) { ... }'. (OK)
                    //
                    // In the first case, 'wakenUp' is set to true and the
                    // following 'selector.select(...)' will wake up immediately.
                    // Until 'wakenUp' is set to false again in the next round,
                    // 'wakenUp.compareAndSet(false, true)' will fail, and therefore
                    // any attempt to wake up the Selector will fail, too, causing
                    // the following 'selector.select(...)' call to block
                    // unnecessarily.
                    //
                    // To fix this problem, we wake up the selector again if wakenUp
                    // is true immediately after selector.select(...).
                    // It is inefficient in that it wakes up the selector for both
                    // the first case (BAD - wake-up required) and the second case
                    // (OK - no wake-up required).

                    if (wakenUp.get()) {
                        selector.wakeup();
                    }
                default:
                    // fallthrough
            }

            cancelledKeys = 0;
            needsToSelectAgain = false;
            final int ioRatio = this.ioRatio;
            if (ioRatio == 100) {
                try {
                    processSelectedKeys();
                } finally {
                    // Ensure we always run tasks.
                    //会执行init方法中的ch.eventLoop().execute
                    runAllTasks();
                }
            } else {
                final long ioStartTime = System.nanoTime();
                try {
                    processSelectedKeys();
                } finally {
                    // Ensure we always run tasks.
                    final long ioTime = System.nanoTime() - ioStartTime;
                    runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                }
            }
        } catch (Throwable t) {
            handleLoopException(t);
        }
        // Always handle shutdown even if the loop processing threw an exception.
        try {
            if (isShuttingDown()) {
                closeAll();
                if (confirmShutdown()) {
                    return;
                }
            }
        } catch (Throwable t) {
            handleLoopException(t);
        }
    }
}

此时可以看到nioeventloop 就是处理socket时间以及处理nioeventloop添加的任务，
看下runtask里面做了啥，前面看下register里面的eventLoop.execute添加了一个任务，即register0。我们可以看下register0干了啥

scss复制代码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();
        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.
        //此处是处理管道添加的所有hanler,在register方法中，服务端管道添加了一个new ChannelInitializer<Channel>，这里面追执行启动代码的handler即TestHander
        pipeline.invokeHandlerAddedIfNeeded();

        safeSetSuccess(promise);
        //调用下一个handler
        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);
    }
}

TestHander如下：

java复制代码public class TestHander implements ChannelHandler {
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        System.out.println();

    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        System.out.println();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println();
    }
}

最后服务端通道只有一个channelHandle没执行，就是ServerBootstrapAcceptor

这个在有socket事件的时候才会触发

scss复制代码private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
    if (!k.isValid()) {
        final EventLoop eventLoop;
        try {
            eventLoop = ch.eventLoop();
        } catch (Throwable ignored) {
            // If the channel implementation throws an exception because there is no event loop, we ignore this
            // because we are only trying to determine if ch is registered to this event loop and thus has authority
            // to close ch.
            return;
        }
        // Only close ch if ch is still registerd to this EventLoop. ch could have deregistered from the event loop
        // and thus the SelectionKey could be cancelled as part of the deregistration process, but the channel is
        // still healthy and should not be closed.
        // See https://github.com/netty/netty/issues/5125
        if (eventLoop != this || eventLoop == null) {
            return;
        }
        // close the channel if the key is not valid anymore
        unsafe.close(unsafe.voidPromise());
        return;
    }

    try {
        int readyOps = k.readyOps();
        // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
        // the NIO JDK channel implementation may throw a NotYetConnectedException.
        if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
            // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
            // See https://github.com/netty/netty/issues/924
            int ops = k.interestOps();
            ops &= ~SelectionKey.OP_CONNECT;
            k.interestOps(ops);

            unsafe.finishConnect();
        }

        // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
        if ((readyOps & SelectionKey.OP_WRITE) != 0) {
            // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
            ch.unsafe().forceFlush();
        }

        // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
        // to a spin loop
        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
        //监听事件，当时连接时间的时候，会执行ServerBootstrapAcceptor的read方法
            unsafe.read();
            if (!ch.isOpen()) {
                // Connection already closed - no need to handle write.
                return;
            }
        }
    } catch (CancelledKeyException ignored) {
        unsafe.close(unsafe.voidPromise());
    }
}

3. 服务端channel绑定address

上面分析完后，最后服务端绑定sserversocket即可。

arduino复制代码private static void doBind0(
        final ChannelFuture regFuture, final Channel channel,
        final SocketAddress localAddress, final ChannelPromise promise) {

    // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
    // the pipeline in its channelRegistered() implementation.
    //这个回调也是eventloop的线程池执行的，线程池就一个线程fastThread
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (regFuture.isSuccess()) {
                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                promise.setFailure(regFuture.cause());
            }
        }
    });
}

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有