reactor-netty中TcpClient的create過程

序

本文主要研究一下reactor-netty中TcpClient的create的過程

maven

<dependency>
            <groupId>io.projectreactor.ipc</groupId>
            <artifactId>reactor-netty</artifactId>
            <version>0.7.3.RELEASE</version>
        </dependency>

TcpClient

reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/tcp/TcpClient.java

protected TcpClient(TcpClient.Builder builder) {
        ClientOptions.Builder<?> clientOptionsBuilder = ClientOptions.builder();
        if (Objects.nonNull(builder.options)) {
            builder.options.accept(clientOptionsBuilder);
        }
        if (!clientOptionsBuilder.isLoopAvailable()) {
            clientOptionsBuilder.loopResources(TcpResources.get());
        }
        if (!clientOptionsBuilder.isPoolAvailable() && !clientOptionsBuilder.isPoolDisabled()) {
            clientOptionsBuilder.poolResources(TcpResources.get());
        }
        this.options = clientOptionsBuilder.build();
    }

loopResources和poolResources實際上是經過TcpResources建立
上面loopResources建立完以後，下面的poolResources實際上是直接返回

clientOptionsBuilder.isLoopAvailable()

reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/options/NettyOptions.java

public final boolean isLoopAvailable() {
            return this.loopResources != null;
        }

一開始是null，因而調用TcpResources.get()建立

TcpResources.get()

reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/tcp/TcpResources.java

/**
     * Return the global HTTP resources for event loops and pooling
     *
     * @return the global HTTP resources for event loops and pooling
     */
    public static TcpResources get() {
        return getOrCreate(tcpResources, null, null, ON_TCP_NEW,  "tcp");
    }
    /**
     * Safely check if existing resource exist and proceed to update/cleanup if new
     * resources references are passed.
     *
     * @param ref the resources atomic reference
     * @param loops the eventual new {@link LoopResources}
     * @param pools the eventual new {@link PoolResources}
     * @param onNew a {@link TcpResources} factory
     * @param name a name for resources
     * @param <T> the reified type of {@link TcpResources}
     *
     * @return an existing or new {@link TcpResources}
     */
    protected static <T extends TcpResources> T getOrCreate(AtomicReference<T> ref,
            LoopResources loops,
            PoolResources pools,
            BiFunction<LoopResources, PoolResources, T> onNew,
            String name) {
        T update;
        for (; ; ) {
            T resources = ref.get();
            if (resources == null || loops != null || pools != null) {
                update = create(resources, loops, pools, name, onNew);
                if (ref.compareAndSet(resources, update)) {
                    if(resources != null){
                        if(loops != null){
                            resources.defaultLoops.dispose();
                        }
                        if(pools != null){
                            resources.defaultPools.dispose();
                        }
                    }
                    return update;
                }
                else {
                    update._dispose();
                }
            }
            else {
                return resources;
            }
        }
    }

這裏進入create，建立loops還有pools

static final AtomicReference<TcpResources>                          tcpResources;
    static final BiFunction<LoopResources, PoolResources, TcpResources> ON_TCP_NEW;

    static {
        ON_TCP_NEW = TcpResources::new;
        tcpResources  = new AtomicReference<>();
    }

    final PoolResources defaultPools;
    final LoopResources defaultLoops;

    protected TcpResources(LoopResources defaultLoops, PoolResources defaultPools) {
        this.defaultLoops = defaultLoops;
        this.defaultPools = defaultPools;
    }

    static <T extends TcpResources> T create(T previous,
            LoopResources loops,
            PoolResources pools,
            String name,
            BiFunction<LoopResources, PoolResources, T> onNew) {
        if (previous == null) {
            loops = loops == null ? LoopResources.create("reactor-" + name) : loops;
            pools = pools == null ? PoolResources.elastic(name) : pools;
        }
        else {
            loops = loops == null ? previous.defaultLoops : loops;
            pools = pools == null ? previous.defaultPools : pools;
        }
        return onNew.apply(loops, pools);
    }

這裏的onNew是建立TcpResources，使用的構造器是TcpResources(LoopResources defaultLoops, PoolResources defaultPools)

LoopResources.create

reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/resources/LoopResources.java

/**
     * Default worker thread count, fallback to available processor
     */
    int DEFAULT_IO_WORKER_COUNT = Integer.parseInt(System.getProperty(
            "reactor.ipc.netty.workerCount",
            "" + Math.max(Runtime.getRuntime()
                        .availableProcessors(), 4)));
    /**
     * Default selector thread count, fallback to -1 (no selector thread)
     */
    int DEFAULT_IO_SELECT_COUNT = Integer.parseInt(System.getProperty(
            "reactor.ipc.netty.selectCount",
            "" + -1));
    /**
     * Create a simple {@link LoopResources} to provide automatically for {@link
     * EventLoopGroup} and {@link Channel} factories
     *
     * @param prefix the event loop thread name prefix
     *
     * @return a new {@link LoopResources} to provide automatically for {@link
     * EventLoopGroup} and {@link Channel} factories
     */
    static LoopResources create(String prefix) {
        return new DefaultLoopResources(prefix, DEFAULT_IO_SELECT_COUNT,
                DEFAULT_IO_WORKER_COUNT,
                true);
    }

這裏有兩個參數，一個是worker thread count，一個是selector thread count

• DEFAULT_IO_WORKER_COUNT

若是環境變量有設置reactor.ipc.netty.workerCount，則用該值；沒有設置則取Math.max(Runtime.getRuntime().availableProcessors(), 4)))

• DEFAULT_IO_SELECT_COUNT

若是環境變量有設置reactor.ipc.netty.selectCount，則用該值；沒有設置則取-1，表示沒有selector thread

DefaultLoopResources

reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/resources/DefaultLoopResources.java

DefaultLoopResources(String prefix,
            int selectCount,
            int workerCount,
            boolean daemon) {
        this.running = new AtomicBoolean(true);
        this.daemon = daemon;
        this.workerCount = workerCount;
        this.prefix = prefix;

        this.serverLoops = new NioEventLoopGroup(workerCount,
                threadFactory(this, "nio"));

        this.clientLoops = LoopResources.colocate(serverLoops);

        this.cacheNativeClientLoops = new AtomicReference<>();
        this.cacheNativeServerLoops = new AtomicReference<>();

        if (selectCount == -1) {
            this.selectCount = workerCount;
            this.serverSelectLoops = this.serverLoops;
            this.cacheNativeSelectLoops = this.cacheNativeServerLoops;
        }
        else {
            this.selectCount = selectCount;
            this.serverSelectLoops =
                    new NioEventLoopGroup(selectCount, threadFactory(this, "select-nio"));
            this.cacheNativeSelectLoops = new AtomicReference<>();
        }
    }

這裏prefix爲reactor-tcp，selectCount爲-1，workerCount爲4，daemon爲true
能夠看到這裏建立了NioEventLoopGroup，workerCount爲4；因爲selectCount=-1所以沒有單首創建serverSelectLoops

NioEventLoopGroup

netty-transport-4.1.20.Final-sources.jar!/io/netty/channel/nio/NioEventLoopGroup.java

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

注意這裏的rejectHandler是RejectedExecutionHandlers.reject()

netty-common-4.1.20.Final-sources.jar!/io/netty/util/concurrent/MultithreadEventExecutorGroup.java

/**
     * Create a new instance.
     *
     * @param nThreads          the number of threads that will be used by this instance.
     * @param threadFactory     the ThreadFactory 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, ThreadFactory threadFactory, Object... args) {
        this(nThreads, threadFactory == null ? null : new ThreadPerTaskExecutor(threadFactory), args);
    }

new NioEventLoopGroup的時候調用了MultithreadEventExecutorGroup
這裏的threadFactory是reactor.ipc.netty.resources.DefaultLoopResources$EventLoopSelectorFactory
這裏的executor是ThreadPerTaskExecutor

netty-common-4.1.20.Final-sources.jar!/io/netty/util/concurrent/ThreadPerTaskExecutor.java

public final class ThreadPerTaskExecutor implements Executor {
    private final ThreadFactory threadFactory;

    public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
        if (threadFactory == null) {
            throw new NullPointerException("threadFactory");
        }
        this.threadFactory = threadFactory;
    }

    @Override
    public void execute(Runnable command) {
        threadFactory.newThread(command).start();
    }
}

MultithreadEventExecutorGroup

/**
     * 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 chooserFactory    the {@link EventExecutorChooserFactory} to use.
     * @param args              arguments which will passed to each {@link #newChild(Executor, Object...)} call
     */
    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) {
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
        }

        children = new EventExecutor[nThreads];

        for (int i = 0; i < nThreads; i ++) {
            boolean success = false;
            try {
                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;
                        }
                    }
                }
            }
        }

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

注意，這裏用for循環去newChild

netty-transport-4.1.20.Final-sources.jar!/io/netty/channel/nio/NioEventLoopGroup.java

protected EventLoop newChild(Executor executor, Object... args) throws Exception {
        return new NioEventLoop(this, executor, (SelectorProvider) args[0],
            ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
    }

每一個child都是一個NioEventLoop

NioEventLoop

netty-transport-4.1.20.Final-sources.jar!/io/netty/channel/nio/NioEventLoop.java

protected static final int DEFAULT_MAX_PENDING_TASKS = Math.max(16,
            SystemPropertyUtil.getInt("io.netty.eventLoop.maxPendingTasks", Integer.MAX_VALUE));

    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;
        final SelectorTuple selectorTuple = openSelector();
        selector = selectorTuple.selector;
        unwrappedSelector = selectorTuple.unwrappedSelector;
        selectStrategy = strategy;
    }

注意這裏的DEFAULT_MAX_PENDING_TASKS參數，指定了隊列的大小。
若是io.netty.eventLoop.maxPendingTasks有指定，則取它跟16的最大值；沒有指定則是Integer.MAX_VALUE
這裏沒有指定，默認是Integer.MAX_VALUE

NioEventLoop extends SingleThreadEventLoop

netty-transport-4.1.20.Final-sources.jar!/io/netty/channel/SingleThreadEventLoop.java

protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,
                                    boolean addTaskWakesUp, int maxPendingTasks,
                                    RejectedExecutionHandler rejectedExecutionHandler) {
        super(parent, executor, addTaskWakesUp, maxPendingTasks, rejectedExecutionHandler);
        tailTasks = newTaskQueue(maxPendingTasks);
    }

這裏的parent是NioEventLoopGroup
這裏的executor是ThreadPerTaskExecutor
這裏的rejectHandler是RejectedExecutionHandlers.reject()

SingleThreadEventLoop extends SingleThreadEventExecutor

/**
     * Create a new instance
     *
     * @param parent            the {@link EventExecutorGroup} which is the parent of this instance and belongs to it
     * @param executor          the {@link Executor} which will be used for executing
     * @param addTaskWakesUp    {@code true} if and only if invocation of {@link #addTask(Runnable)} will wake up the
     *                          executor thread
     * @param maxPendingTasks   the maximum number of pending tasks before new tasks will be rejected.
     * @param rejectedHandler   the {@link RejectedExecutionHandler} to use.
     */
    protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
                                        boolean addTaskWakesUp, int maxPendingTasks,
                                        RejectedExecutionHandler rejectedHandler) {
        super(parent);
        this.addTaskWakesUp = addTaskWakesUp;
        this.maxPendingTasks = Math.max(16, maxPendingTasks);
        this.executor = ObjectUtil.checkNotNull(executor, "executor");
        taskQueue = newTaskQueue(this.maxPendingTasks);
        rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
    }

    /**
     * Create a new {@link Queue} which will holds the tasks to execute. This default implementation will return a
     * {@link LinkedBlockingQueue} but if your sub-class of {@link SingleThreadEventExecutor} will not do any blocking
     * calls on the this {@link Queue} it may make sense to {@code @Override} this and return some more performant
     * implementation that does not support blocking operations at all.
     */
    protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
        return new LinkedBlockingQueue<Runnable>(maxPendingTasks);
    }

這裏的maxPendingTasks是Integer.MAX_VALUE，建立的taskQueue的大小爲Integer.MAX_VALUE
這裏的addTaskWakesUp爲false

PoolResources.elastic(name)

reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/resources/PoolResources.java

/**
     * Create an uncapped {@link PoolResources} to provide automatically for {@link
     * ChannelPool}.
     * <p>An elastic {@link PoolResources} will never wait before opening a new
     * connection. The reuse window is limited but it cannot starve an undetermined volume
     * of clients using it.
     *
     * @param name the channel pool map name
     *
     * @return a new {@link PoolResources} to provide automatically for {@link
     * ChannelPool}
     */
    static PoolResources elastic(String name) {
        return new DefaultPoolResources(name, SimpleChannelPool::new);
    }

DefaultPoolResources

reactor-netty-0.7.3.RELEASE-sources.jar!/reactor/ipc/netty/resources/DefaultPoolResources.java

final ConcurrentMap<SocketAddress, Pool> channelPools;
    final String                             name;
    final PoolFactory                        provider;

    DefaultPoolResources(String name, PoolFactory provider) {
        this.name = name;
        this.provider = provider;
        this.channelPools = PlatformDependent.newConcurrentHashMap();
    }

建立channelPools的map，key是SocketAddress，value是Pool

小結

TcpClient的create方法主要是建立TcpResources，而TcpResources則建立loopResources和poolResources。

loopResources

這個loopResources主要是建立NioEventLoopGroup，以及該group下面的workerCount個NioEventLoop(這裏涉及兩個參數，一個是worker thread count，一個是selector thread count)

• DEFAULT_IO_WORKER_COUNT：若是環境變量有設置reactor.ipc.netty.workerCount，則用該值；沒有設置則取Math.max(Runtime.getRuntime().availableProcessors(), 4)))
• DEFAULT_IO_SELECT_COUNT：若是環境變量有設置reactor.ipc.netty.selectCount，則用該值；沒有設置則取-1，表示沒有selector thread
• DEFAULT_MAX_PENDING_TASKS: 指定NioEventLoop的taskQueue的大小，Math.max(16,SystemPropertyUtil.getInt("io.netty.eventLoop.maxPendingTasks", Integer.MAX_VALUE))
• NioEventLoop繼承了SingleThreadEventLoop，而SingleThreadEventLoop則繼承SingleThreadEventExecutor，而其代理的executor是ThreadPerTaskExecutor，rejectHandler是RejectedExecutionHandlers.reject()，默認的taskQueue是LinkedBlockingQueue，其大小爲Integer.MAX_VALUE

poolResources

這個主要是建立channelPools，類型是ConcurrentMap<SocketAddress, Pool>