聊聊 Netty 那些事儿之 Reactor 在 Netty 中的实现 (创建篇)

/** * Echoes back any received data from a client. */public final class EchoServer {    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
    public static void main(String[] args) throws Exception {        // Configure the server.        //创建主从Reactor线程组        EventLoopGroup bossGroup = new NioEventLoopGroup(1);        EventLoopGroup workerGroup = new NioEventLoopGroup();        final EchoServerHandler serverHandler = new EchoServerHandler();        try {            ServerBootstrap b = new ServerBootstrap();            b.group(bossGroup, workerGroup)//配置主从Reactor             .channel(NioServerSocketChannel.class)//配置主Reactor中的channel类型             .option(ChannelOption.SO_BACKLOG, 100)//设置主Reactor中channel的option选项             .handler(new LoggingHandler(LogLevel.INFO))//设置主Reactor中Channel->pipline->handler             .childHandler(new ChannelInitializer<SocketChannel>() {//设置从Reactor中注册channel的pipeline                 @Override                 public void initChannel(SocketChannel ch) throws Exception {                     ChannelPipeline p = ch.pipeline();                     //p.addLast(new LoggingHandler(LogLevel.INFO));                     p.addLast(serverHandler);                 }             });
            // Start the server. 绑定端口启动服务，开始监听accept事件            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();        }    }}

  //创建主从Reactor线程组  EventLoopGroup bossGroup = new NioEventLoopGroup(1);  EventLoopGroup workerGroup = new NioEventLoopGroup();

final EchoServerHandler serverHandler = new EchoServerHandler();
/** * Handler implementation for the echo server. */@Sharablepublic class EchoServerHandler extends ChannelInboundHandlerAdapter {
    @Override    public void channelRead(ChannelHandlerContext ctx, Object msg) {        ................省略IO处理逻辑................        ctx.write(msg);    }
    @Override    public void channelReadComplete(ChannelHandlerContext ctx) {                ctx.flush();    }
    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {        // Close the connection when an exception is raised.        cause.printStackTrace();        ctx.close();    }}

  //创建主从Reactor线程组  EventLoopGroup bossGroup = new NioEventLoopGroup(1);  EventLoopGroup workerGroup = new NioEventLoopGroup();

public class NioEventLoopGroup extends MultithreadEventLoopGroup {
    /**     * Create a new instance using the default number of threads, the default {@link ThreadFactory} and     * the {@link SelectorProvider} which is returned by {@link SelectorProvider#provider()}.     */    public NioEventLoopGroup() {        this(0);    }
    /**     * Create a new instance using the specified number of threads, {@link ThreadFactory} and the     * {@link SelectorProvider} which is returned by {@link SelectorProvider#provider()}.     */    public NioEventLoopGroup(int nThreads) {        this(nThreads, (Executor) null);    }
    ......................省略...........................}

    public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,                             final SelectStrategyFactory selectStrategyFactory) {        super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());    }

public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(MultithreadEventLoopGroup.class);    //默认Reactor个数    private static final int DEFAULT_EVENT_LOOP_THREADS;
    static {        DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(                "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
        if (logger.isDebugEnabled()) {            logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS);        }    }
    /**     * @see MultithreadEventExecutorGroup#MultithreadEventExecutorGroup(int, Executor, Object...)     */    protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);    }
    ...................省略.....................}

public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
    //Reactor线程组中的Reactor集合    private final EventExecutor[] children;    private final Set<EventExecutor> readonlyChildren;    //从Reactor group中选择一个特定的Reactor的选择策略 用于channel注册绑定到一个固定的Reactor上    private final EventExecutorChooserFactory.EventExecutorChooser chooser;
    /**     * Create a new instance.     *     * @param nThreads          the number of threads that will be used by this instance.     * @param executor          the Executor to use, or {@code null} if the default should be used.     * @param args              arguments which will passed to each {@link #newChild(Executor, Object...)} call     */    protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {        this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);    }
    ............................省略................................}

    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) {            //用于创建Reactor线程            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());        }
        children = new EventExecutor[nThreads];        //循环创建reaactor group中的Reactor        for (int i = 0; i < nThreads; i ++) {            boolean success = false;            try {                //创建reactor                children[i] = newChild(executor, args);                success = true;            } catch (Exception e) {                throw new IllegalStateException("failed to create a child event loop", e);            } finally {                     ................省略................                }            }        }        //创建channel到Reactor的绑定策略        chooser = chooserFactory.newChooser(children);
         ................省略................
        Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);        Collections.addAll(childrenSet, children);        readonlyChildren = Collections.unmodifiableSet(childrenSet);    }

public final class ThreadPerTaskExecutor implements Executor {    private final ThreadFactory threadFactory;
    public ThreadPerTaskExecutor(ThreadFactory threadFactory) {        this.threadFactory = ObjectUtil.checkNotNull(threadFactory, "threadFactory");    }
    @Override    public void execute(Runnable command) {        threadFactory.newThread(command).start();    }}

protected abstract EventExecutor newChild(Executor executor, Object... args) throws Exception;

public class NioEventLoopGroup extends MultithreadEventLoopGroup {    @Override    protected EventLoop newChild(Executor executor, Object... args) throws Exception {        EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null;        return new NioEventLoop(this, executor, (SelectorProvider) args[0],            ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2], queueFactory);    }}

public final class NioEventLoop extends SingleThreadEventLoop {    //用于创建JDK NIO Selector,ServerSocketChannel    private final SelectorProvider provider;    //Selector轮询策略 决定什么时候轮询，什么时候处理IO事件，什么时候执行异步任务    private final SelectStrategy selectStrategy;    /**     * The NIO {@link Selector}.     */    private Selector selector;    private Selector unwrappedSelector;
    NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,                 SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,                 EventLoopTaskQueueFactory queueFactory) {        super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),                rejectedExecutionHandler);        this.provider = ObjectUtil.checkNotNull(selectorProvider, "selectorProvider");        this.selectStrategy = ObjectUtil.checkNotNull(strategy, "selectStrategy");        final SelectorTuple selectorTuple = openSelector();        this.selector = selectorTuple.selector;        this.unwrappedSelector = selectorTuple.unwrappedSelector;    }}

 private SelectorTuple openSelector() {        final Selector unwrappedSelector;        try {            //通过JDK NIO SelectorProvider创建Selector            unwrappedSelector = provider.openSelector();        } catch (IOException e) {            throw new ChannelException("failed to open a new selector", e);        }
        ..................省略.............}

    public NioEventLoopGroup(ThreadFactory threadFactory) {        this(0, threadFactory, SelectorProvider.provider());    }

public abstract class SelectorProvider {
    public static SelectorProvider provider() {        synchronized (lock) {            if (provider != null)                return provider;            return AccessController.doPrivileged(                new PrivilegedAction<SelectorProvider>() {                    public SelectorProvider run() {                            if (loadProviderFromProperty())                                return provider;                            if (loadProviderAsService())                                return provider;                            provider = sun.nio.ch.DefaultSelectorProvider.create();                            return provider;                        }                    });        }    }}

    private static boolean loadProviderFromProperty() {        String cn = System.getProperty("java.nio.channels.spi.SelectorProvider");        if (cn == null)            return false;        try {            Class<?> c = Class.forName(cn, true,                                       ClassLoader.getSystemClassLoader());            provider = (SelectorProvider)c.newInstance();            return true;        }         .................省略.............    }

    private static boolean loadProviderAsService() {
        ServiceLoader<SelectorProvider> sl =            ServiceLoader.load(SelectorProvider.class,                               ClassLoader.getSystemClassLoader());        Iterator<SelectorProvider> i = sl.iterator();        for (;;) {            try {                if (!i.hasNext())                    return false;                provider = i.next();                return true;            } catch (ServiceConfigurationError sce) {                if (sce.getCause() instanceof SecurityException) {                    // Ignore the security exception, try the next provider                    continue;                }                throw sce;            }        }    }

public class DefaultSelectorProvider {    private DefaultSelectorProvider() {    }
    public static SelectorProvider create() {        return new KQueueSelectorProvider();    }}

public final class NioEventLoop extends SingleThreadEventLoop {   //Selector优化开关 默认开启 为了遍历的效率 会对Selector中的SelectedKeys进行数据结构优化    private static final boolean DISABLE_KEY_SET_OPTIMIZATION =            SystemPropertyUtil.getBoolean("io.netty.noKeySetOptimization", false);}

private SelectorTuple openSelector() {        ..........SelectorProvider创建JDK NIO  原生Selector..............
        if (DISABLE_KEY_SET_OPTIMIZATION) {            //JDK NIO原生Selector ，Selector优化开关 默认开启需要对Selector进行优化            return new SelectorTuple(unwrappedSelector);        }}

       Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() {            @Override            public Object run() {                try {                    return Class.forName(                            "sun.nio.ch.SelectorImpl",                            false,                            PlatformDependent.getSystemClassLoader());                } catch (Throwable cause) {                    return cause;                }            }        });

public abstract class SelectorImpl extends AbstractSelector {
    // The set of keys with data ready for an operation    // //IO就绪的SelectionKey（里面包裹着channel）    protected Set<SelectionKey> selectedKeys;
    // The set of keys registered with this Selector    //注册在该Selector上的所有SelectionKey（里面包裹着channel）    protected HashSet<SelectionKey> keys;
    // Public views of the key sets    //用于向调用线程返回的keys，不可变    private Set<SelectionKey> publicKeys;             // Immutable    //当有IO就绪的SelectionKey时，向调用线程返回。只可删除其中元素，不可增加    private Set<SelectionKey> publicSelectedKeys;     // Removal allowed, but not addition
    protected SelectorImpl(SelectorProvider sp) {        super(sp);        keys = new HashSet<SelectionKey>();        selectedKeys = new HashSet<SelectionKey>();        if (Util.atBugLevel("1.4")) {            publicKeys = keys;            publicSelectedKeys = selectedKeys;        } else {            //不可变            publicKeys = Collections.unmodifiableSet(keys);            //只可删除其中元素，不可增加            publicSelectedKeys = Util.ungrowableSet(selectedKeys);        }    }}

        //判断是否可以对Selector进行优化，这里主要针对JDK NIO原生Selector的实现类进行优化，因为SelectorProvider可以加载的是自定义Selector实现        //如果SelectorProvider创建的Selector不是JDK原生sun.nio.ch.SelectorImpl的实现类，那么无法进行优化，直接返回        if (!(maybeSelectorImplClass instanceof Class) ||            !((Class<?>) maybeSelectorImplClass).isAssignableFrom(unwrappedSelector.getClass())) {            if (maybeSelectorImplClass instanceof Throwable) {                Throwable t = (Throwable) maybeSelectorImplClass;                logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, t);            }            return new SelectorTuple(unwrappedSelector);        }

class KQueueSelectorImpl extends SelectorImpl {
}

final class SelectedSelectionKeySet extends AbstractSet<SelectionKey> {
    //采用数组替换到JDK中的HashSet,这样add操作和遍历操作效率更高，不需要考虑hash冲突    SelectionKey[] keys;    //数组尾部指针    int size;
    SelectedSelectionKeySet() {        keys = new SelectionKey[1024];    }
    /**     * 数组的添加效率高于 HashSet 因为不需要考虑hash冲突     * */    @Override    public boolean add(SelectionKey o) {        if (o == null) {            return false;        }        //时间复杂度O（1）        keys[size++] = o;        if (size == keys.length) {            //扩容为原来的两倍大小            increaseCapacity();        }
        return true;    }
    private void increaseCapacity() {        SelectionKey[] newKeys = new SelectionKey[keys.length << 1];        System.arraycopy(keys, 0, newKeys, 0, size);        keys = newKeys;    }
    /**     * 采用数组的遍历效率 高于 HashSet     * */    @Override    public Iterator<SelectionKey> iterator() {        return new Iterator<SelectionKey>() {            private int idx;
            @Override            public boolean hasNext() {                return idx < size;            }
            @Override            public SelectionKey next() {                if (!hasNext()) {                    throw new NoSuchElementException();                }                return keys[idx++];            }
            @Override            public void remove() {                throw new UnsupportedOperationException();            }        };    }}

  Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys");  Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys");

       if (PlatformDependent.javaVersion() >= 9 && PlatformDependent.hasUnsafe()) {
                        long selectedKeysFieldOffset = PlatformDependent.objectFieldOffset(selectedKeysField);                        long publicSelectedKeysFieldOffset =                                PlatformDependent.objectFieldOffset(publicSelectedKeysField);
                        if (selectedKeysFieldOffset != -1 && publicSelectedKeysFieldOffset != -1) {                            PlatformDependent.putObject(                                    unwrappedSelector, selectedKeysFieldOffset, selectedKeySet);                            PlatformDependent.putObject(                                    unwrappedSelector, publicSelectedKeysFieldOffset, selectedKeySet);                            return null;                        }                                            }

          Throwable cause = ReflectionUtil.trySetAccessible(selectedKeysField, true);          if (cause != null) {                return cause;          }          cause = ReflectionUtil.trySetAccessible(publicSelectedKeysField, true);          if (cause != null) {                return cause;          }          //Java8反射替换字段          selectedKeysField.set(unwrappedSelector, selectedKeySet);          publicSelectedKeysField.set(unwrappedSelector, selectedKeySet);

   //会通过反射替换selector对象中的selectedKeySet保存就绪的selectKey    //该字段为持有selector对象selectedKeys的引用，当IO事件就绪时，直接从这里获取    private SelectedSelectionKeySet selectedKeys;

return new SelectorTuple(unwrappedSelector,                      new SelectedSelectionKeySetSelector(unwrappedSelector, selectedKeySet));

    private static final class SelectorTuple {        final Selector unwrappedSelector;        final Selector selector;
        SelectorTuple(Selector unwrappedSelector) {            this.unwrappedSelector = unwrappedSelector;            this.selector = unwrappedSelector;        }
        SelectorTuple(Selector unwrappedSelector, Selector selector) {            this.unwrappedSelector = unwrappedSelector;            this.selector = selector;        }    }

final class SelectedSelectionKeySetSelector extends Selector {    //Netty优化后的 SelectedKey就绪集合    private final SelectedSelectionKeySet selectionKeys;    //优化后的JDK NIO 原生Selector    private final Selector delegate;
    SelectedSelectionKeySetSelector(Selector delegate, SelectedSelectionKeySet selectionKeys) {        this.delegate = delegate;        this.selectionKeys = selectionKeys;    }
    @Override    public boolean isOpen() {        return delegate.isOpen();    }
    @Override    public SelectorProvider provider() {        return delegate.provider();    }
    @Override    public Set<SelectionKey> keys() {        return delegate.keys();    }
    @Override    public Set<SelectionKey> selectedKeys() {        return delegate.selectedKeys();    }
    @Override    public int selectNow() throws IOException {        //重置SelectedKeys集合        selectionKeys.reset();        return delegate.selectNow();    }
    @Override    public int select(long timeout) throws IOException {        //重置SelectedKeys集合        selectionKeys.reset();        return delegate.select(timeout);    }
    @Override    public int select() throws IOException {        //重置SelectedKeys集合        selectionKeys.reset();        return delegate.select();    }
    @Override    public Selector wakeup() {        return delegate.wakeup();    }
    @Override    public void close() throws IOException {        delegate.close();    }}

    NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,                 SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,                 EventLoopTaskQueueFactory queueFactory) {        super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),                rejectedExecutionHandler);        this.provider = ObjectUtil.checkNotNull(selectorProvider, "selectorProvider");        this.selectStrategy = ObjectUtil.checkNotNull(strategy, "selectStrategy");        final SelectorTuple selectorTuple = openSelector();        //通过用SelectedSelectionKeySet装饰后的unwrappedSelector        this.selector = selectorTuple.selector;        //Netty优化过的JDK NIO远程Selector        this.unwrappedSelector = selectorTuple.unwrappedSelector;    }

    //任务队列大小，默认是无界队列    protected static final int DEFAULT_MAX_PENDING_TASKS = Math.max(16,            SystemPropertyUtil.getInt("io.netty.eventLoop.maxPendingTasks", Integer.MAX_VALUE));
    private static Queue<Runnable> newTaskQueue(            EventLoopTaskQueueFactory queueFactory) {        if (queueFactory == null) {            return newTaskQueue0(DEFAULT_MAX_PENDING_TASKS);        }        return queueFactory.newTaskQueue(DEFAULT_MAX_PENDING_TASKS);    }
    private static Queue<Runnable> newTaskQueue0(int maxPendingTasks) {        // This event loop never calls takeTask()        return maxPendingTasks == Integer.MAX_VALUE ? PlatformDependent.<Runnable>newMpscQueue()                : PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks);    }  

    //创建无界任务队列    PlatformDependent.<Runnable>newMpscQueue()    //创建有界任务队列    PlatformDependent.<Runnable>newMpscQueue(maxPendingTasks)
    public static <T> Queue<T> newMpscQueue() {        return Mpsc.newMpscQueue();    }
    public static <T> Queue<T> newMpscQueue(final int maxCapacity) {        return Mpsc.newMpscQueue(maxCapacity);    }

public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {
    //任务队列大小，默认是无界队列    protected static final int DEFAULT_MAX_PENDING_TASKS = Math.max(16,            SystemPropertyUtil.getInt("io.netty.eventLoop.maxPendingTasks", Integer.MAX_VALUE));        //尾部任务队列    private final Queue<Runnable> tailTasks;
    protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,                                    boolean addTaskWakesUp, Queue<Runnable> taskQueue, Queue<Runnable> tailTaskQueue,                                    RejectedExecutionHandler rejectedExecutionHandler) {        super(parent, executor, addTaskWakesUp, taskQueue, rejectedExecutionHandler);        //尾部队列 执行一些统计任务 不常用        tailTasks = ObjectUtil.checkNotNull(tailTaskQueue, "tailTaskQueue");    }
    @Override    public ChannelFuture register(Channel channel) {        //注册channel到绑定的Reactor上        return register(new DefaultChannelPromise(channel, this));    }}

public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
    protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,                                        boolean addTaskWakesUp, Queue<Runnable> taskQueue, RejectedExecutionHandler rejectedHandler) {        //parent为Reactor所属的NioEventLoopGroup Reactor线程组        super(parent);        //向Reactor添加任务时，是否唤醒Selector停止轮询IO就绪事件，马上执行异步任务        this.addTaskWakesUp = addTaskWakesUp;        //Reactor异步任务队列的大小        this.maxPendingTasks = DEFAULT_MAX_PENDING_EXECUTOR_TASKS;        //用于启动Reactor线程的executor -> ThreadPerTaskExecutor        this.executor = ThreadExecutorMap.apply(executor, this);        //普通任务队列        this.taskQueue = ObjectUtil.checkNotNull(taskQueue, "taskQueue");        //任务队列满时的拒绝策略        this.rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");    }}

public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {   //从Reactor集合中选择一个特定的Reactor的绑定策略 用于channel注册绑定到一个固定的Reactor上    private final EventExecutorChooserFactory.EventExecutorChooser chooser;
    chooser = chooserFactory.newChooser(children);}

public final class DefaultEventExecutorChooserFactory implements EventExecutorChooserFactory {
    public static final DefaultEventExecutorChooserFactory INSTANCE = new DefaultEventExecutorChooserFactory();
    private DefaultEventExecutorChooserFactory() { }
    @Override    public EventExecutorChooser newChooser(EventExecutor[] executors) {        if (isPowerOfTwo(executors.length)) {            return new PowerOfTwoEventExecutorChooser(executors);        } else {            return new GenericEventExecutorChooser(executors);        }    }
    private static boolean isPowerOfTwo(int val) {        return (val & -val) == val;    }    ...................省略.................}

    private static final class GenericEventExecutorChooser implements EventExecutorChooser {        private final AtomicLong idx = new AtomicLong();        private final EventExecutor[] executors;
        GenericEventExecutorChooser(EventExecutor[] executors) {            this.executors = executors;        }
        @Override        public EventExecutor next() {            return executors[(int) Math.abs(idx.getAndIncrement() % executors.length)];        }    }

    private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser {        private final AtomicInteger idx = new AtomicInteger();        private final EventExecutor[] executors;
        PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {            this.executors = executors;        }
        @Override        public EventExecutor next() {            return executors[idx.getAndIncrement() & executors.length - 1];        }    }

       //记录关闭的Reactor个数，当Reactor全部关闭后，才可以认为关闭成功        private final AtomicInteger terminatedChildren = new AtomicInteger();        //关闭future        private final Promise<?> terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE);
        final FutureListener<Object> terminationListener = new FutureListener<Object>() {            @Override            public void operationComplete(Future<Object> future) throws Exception {                if (terminatedChildren.incrementAndGet() == children.length) {                    //当所有Reactor关闭后 才认为是关闭成功                    terminationFuture.setSuccess(null);                }            }        };                //为所有Reactor添加terminationListener        for (EventExecutor e: children) {            e.terminationFuture().addListener(terminationListener);        }

public final class EchoServer {    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
    public static void main(String[] args) throws Exception {        // Configure the server.        //创建主从Reactor线程组        EventLoopGroup bossGroup = new NioEventLoopGroup(1);        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ...........省略............    }}