聊聊reactor-netty的PoolResources的两种模式

        @Override
        public Future<Channel> acquire() {
            return acquire(defaultGroup.next().newPromise());
        }

        @Override
        public Future<Channel> acquire(Promise<Channel> promise) {
            return pool.acquire(promise).addListener(this);
        }

        @Override
        public Future<Void> release(Channel channel) {
            return pool.release(channel);
        }

        @Override
        public Future<Void> release(Channel channel, Promise<Void> promise) {
            return pool.release(channel, promise);
        }

        @Override
        public void close() {
            if(compareAndSet(false, true)) {
                pool.close();
            }
        }

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

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

/**
 * Simple {@link ChannelPool} implementation which will create new {@link Channel}s if someone tries to acquire
 * a {@link Channel} but none is in the pool atm. No limit on the maximal concurrent {@link Channel}s is enforced.
 *
 * This implementation uses LIFO order for {@link Channel}s in the {@link ChannelPool}.
 *
 */
public class SimpleChannelPool implements ChannelPool {

    @Override
    public final Future<Channel> acquire() {
        return acquire(bootstrap.config().group().next().<Channel>newPromise());
    }

    @Override
    public Future<Channel> acquire(final Promise<Channel> promise) {
        checkNotNull(promise, "promise");
        return acquireHealthyFromPoolOrNew(promise);
    }

    /**
     * Tries to retrieve healthy channel from the pool if any or creates a new channel otherwise.
     * @param promise the promise to provide acquire result.
     * @return future for acquiring a channel.
     */
    private Future<Channel> acquireHealthyFromPoolOrNew(final Promise<Channel> promise) {
        try {
            final Channel ch = pollChannel();
            if (ch == null) {
                // No Channel left in the pool bootstrap a new Channel
                Bootstrap bs = bootstrap.clone();
                bs.attr(POOL_KEY, this);
                ChannelFuture f = connectChannel(bs);
                if (f.isDone()) {
                    notifyConnect(f, promise);
                } else {
                    f.addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture future) throws Exception {
                            notifyConnect(future, promise);
                        }
                    });
                }
                return promise;
            }
            EventLoop loop = ch.eventLoop();
            if (loop.inEventLoop()) {
                doHealthCheck(ch, promise);
            } else {
                loop.execute(new Runnable() {
                    @Override
                    public void run() {
                        doHealthCheck(ch, promise);
                    }
                });
            }
        } catch (Throwable cause) {
            promise.tryFailure(cause);
        }
        return promise;
    }

    @Override
    public final Future<Void> release(Channel channel) {
        return release(channel, channel.eventLoop().<Void>newPromise());
    }

    @Override
    public Future<Void> release(final Channel channel, final Promise<Void> promise) {
        checkNotNull(channel, "channel");
        checkNotNull(promise, "promise");
        try {
            EventLoop loop = channel.eventLoop();
            if (loop.inEventLoop()) {
                doReleaseChannel(channel, promise);
            } else {
                loop.execute(new Runnable() {
                    @Override
                    public void run() {
                        doReleaseChannel(channel, promise);
                    }
                });
            }
        } catch (Throwable cause) {
            closeAndFail(channel, cause, promise);
        }
        return promise;
    }

    @Override
    public void close() {
        for (;;) {
            Channel channel = pollChannel();
            if (channel == null) {
                break;
            }
            channel.close();
        }
    }

    //......
}

    /**
     * Default max connection, if -1 will never wait to acquire before opening new
     * connection in an unbounded fashion. Fallback to
     * available number of processors.
     */
    int DEFAULT_POOL_MAX_CONNECTION =
            Integer.parseInt(System.getProperty("reactor.ipc.netty.pool.maxConnections",
            "" + Math.max(Runtime.getRuntime()
                        .availableProcessors(), 8) * 2));

    /**
     * Default acquisition timeout before error. If -1 will never wait to
     * acquire before opening new
     * connection in an unbounded fashion. Fallback to
     * available number of processors.
     */
    long DEFAULT_POOL_ACQUIRE_TIMEOUT = Long.parseLong(System.getProperty(
            "reactor.ipc.netty.pool.acquireTimeout",
            "" + 45000));

    /**
     * Create a capped {@link PoolResources} to provide automatically for {@link
     * ChannelPool}.
     * <p>A Fixed {@link PoolResources} will open up to the given max number of
     * processors observed by this jvm (minimum 4).
     * Further connections will be pending acquisition indefinitely.
     *
     * @param name the channel pool map name
     *
     * @return a new {@link PoolResources} to provide automatically for {@link
     * ChannelPool}
     */
    static PoolResources fixed(String name) {
        return fixed(name, DEFAULT_POOL_MAX_CONNECTION);
    }

    /**
     * Create a capped {@link PoolResources} to provide automatically for {@link
     * ChannelPool}.
     * <p>A Fixed {@link PoolResources} will open up to the given max connection value.
     * Further connections will be pending acquisition indefinitely.
     *
     * @param name the channel pool map name
     * @param maxConnections the maximum number of connections before starting pending
     * acquisition on existing ones
     *
     * @return a new {@link PoolResources} to provide automatically for {@link
     * ChannelPool}
     */
    static PoolResources fixed(String name, int maxConnections) {
        return fixed(name, maxConnections, DEFAULT_POOL_ACQUIRE_TIMEOUT);
    }

    /**
     * Create a capped {@link PoolResources} to provide automatically for {@link
     * ChannelPool}.
     * <p>A Fixed {@link PoolResources} will open up to the given max connection value.
     * Further connections will be pending acquisition indefinitely.
     *
     * @param name the channel pool map name
     * @param maxConnections the maximum number of connections before starting pending
     * @param acquireTimeout the maximum time in millis to wait for aquiring
     *
     * @return a new {@link PoolResources} to provide automatically for {@link
     * ChannelPool}
     */
    static PoolResources fixed(String name, int maxConnections, long acquireTimeout) {
        if (maxConnections == -1) {
            return elastic(name);
        }
        if (maxConnections <= 0) {
            throw new IllegalArgumentException("Max Connections value must be strictly " + "positive");
        }
        if (acquireTimeout != -1L && acquireTimeout < 0) {
            throw new IllegalArgumentException("Acquire Timeout value must " + "be " + "positive");
        }
        return new DefaultPoolResources(name,
                (bootstrap, handler, checker) -> new FixedChannelPool(bootstrap,
                        handler,
                        checker,
                        FixedChannelPool.AcquireTimeoutAction.FAIL,
                        acquireTimeout,
                        maxConnections,
                        Integer.MAX_VALUE
                        ));
    }

/**
 * {@link ChannelPool} implementation that takes another {@link ChannelPool} implementation and enforce a maximum
 * number of concurrent connections.
 */
public class FixedChannelPool extends SimpleChannelPool {

    @Override
    public Future<Channel> acquire(final Promise<Channel> promise) {
        try {
            if (executor.inEventLoop()) {
                acquire0(promise);
            } else {
                executor.execute(new Runnable() {
                    @Override
                    public void run() {
                        acquire0(promise);
                    }
                });
            }
        } catch (Throwable cause) {
            promise.setFailure(cause);
        }
        return promise;
    }

    private void acquire0(final Promise<Channel> promise) {
        assert executor.inEventLoop();

        if (closed) {
            promise.setFailure(POOL_CLOSED_ON_ACQUIRE_EXCEPTION);
            return;
        }
        if (acquiredChannelCount < maxConnections) {
            assert acquiredChannelCount >= 0;

            // We need to create a new promise as we need to ensure the AcquireListener runs in the correct
            // EventLoop
            Promise<Channel> p = executor.newPromise();
            AcquireListener l = new AcquireListener(promise);
            l.acquired();
            p.addListener(l);
            super.acquire(p);
        } else {
            if (pendingAcquireCount >= maxPendingAcquires) {
                promise.setFailure(FULL_EXCEPTION);
            } else {
                AcquireTask task = new AcquireTask(promise);
                if (pendingAcquireQueue.offer(task)) {
                    ++pendingAcquireCount;

                    if (timeoutTask != null) {
                        task.timeoutFuture = executor.schedule(timeoutTask, acquireTimeoutNanos, TimeUnit.NANOSECONDS);
                    }
                } else {
                    promise.setFailure(FULL_EXCEPTION);
                }
            }

            assert pendingAcquireCount > 0;
        }
    }

    @Override
    public Future<Void> release(final Channel channel, final Promise<Void> promise) {
        ObjectUtil.checkNotNull(promise, "promise");
        final Promise<Void> p = executor.newPromise();
        super.release(channel, p.addListener(new FutureListener<Void>() {

            @Override
            public void operationComplete(Future<Void> future) throws Exception {
                assert executor.inEventLoop();

                if (closed) {
                    // Since the pool is closed, we have no choice but to close the channel
                    channel.close();
                    promise.setFailure(POOL_CLOSED_ON_RELEASE_EXCEPTION);
                    return;
                }

                if (future.isSuccess()) {
                    decrementAndRunTaskQueue();
                    promise.setSuccess(null);
                } else {
                    Throwable cause = future.cause();
                    // Check if the exception was not because of we passed the Channel to the wrong pool.
                    if (!(cause instanceof IllegalArgumentException)) {
                        decrementAndRunTaskQueue();
                    }
                    promise.setFailure(future.cause());
                }
            }
        }));
        return promise;
    }

    @Override
    public void close() {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                if (!closed) {
                    closed = true;
                    for (;;) {
                        AcquireTask task = pendingAcquireQueue.poll();
                        if (task == null) {
                            break;
                        }
                        ScheduledFuture<?> f = task.timeoutFuture;
                        if (f != null) {
                            f.cancel(false);
                        }
                        task.promise.setFailure(new ClosedChannelException());
                    }
                    acquiredChannelCount = 0;
                    pendingAcquireCount = 0;
                    FixedChannelPool.super.close();
                }
            }
        });
    }
    //......
}