io.netty.channel.pool.ChannelPoolHandler#io.netty.channel.pool.FixedChannelPool源码实例Demo

/**
 * 释放连接
 * @param channel
 * @return
 */
public Future<Void> release(Channel channel) {
	Verify.verifyNotNull(channel, "channel不允许为NULL");
	InetSocketAddress remoteAddress = (InetSocketAddress)channel.remoteAddress();
	if(logger.isDebugEnabled()) {
		logger.debug("{} channel released", remoteAddress);
	}
	FixedChannelPool fixedChannelPool = pollMap.get(remoteAddress);
	Future<Void> releaseFuture = fixedChannelPool.release(channel);
	if(!releaseFuture.isSuccess()) {
		Throwable cause = releaseFuture.cause();
		if(cause != null) {
			logger.error("rlease local channel {}, remote channel {}, happens error {}", channel.localAddress(),
					channel.remoteAddress(), ExceptionUtils.getStackTrace(releaseFuture.cause()));
		}
	}
	return releaseFuture;
}
 

/**
 * 循环取出可用连接一个使用
 * @return
 * @throws VerifyException
 */
public Future<Channel> acquire(List<ConnectAddrProperty> sas) {
	Verify.verify(pollMap != null && !sas.isEmpty(), "ftdc connection pool init failure or sas is null");
	SocketAddressChooser newChooser = DefaultSocketAddressChooserFactory.INSTANCE.newChooser(sas.size());
	SocketAddressChooser oldChooser = CHOOSER_MAP.putIfAbsent(sas.hashCode(), newChooser);
	if(oldChooser == null) {
		oldChooser = newChooser;
	}
	InetSocketAddress socketAddress = (InetSocketAddress)oldChooser.next(sas);
	if(logger.isDebugEnabled()) {
		logger.debug("use socket address {}", socketAddress);
	}
	FixedChannelPool fixedChannelPool = pollMap.get(socketAddress);
	return fixedChannelPool.acquire();
}
 

private BackClientPool() {
    bootstrap
            .group(group)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 500)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_KEEPALIVE, true);

    poolMap = new AbstractChannelPoolMap<RequestHolder, SimpleChannelPool>() {
        @Override
        protected SimpleChannelPool newPool(RequestHolder requestHolder) {
            return new FixedChannelPool(bootstrap.remoteAddress(requestHolder.getSocketAddress()), new BackPoolHandler(requestHolder), 50);
        }
    };
}
 

private void init(String url, int maxCount) throws URISyntaxException, SSLException {
    URI uri = new URI(url);
    if (uri.getScheme() == null || uri.getHost() == null) {
        throw new IllegalArgumentException("uri不合法");
    }
    scheme = uri.getScheme();
    host = uri.getHost();
    port = uri.getPort();
    if (port == -1) {
        if (HttpScheme.HTTP.equalsIgnoreCase(scheme)) {
            port = 80;
        } else if (HttpScheme.HTTPS.equalsIgnoreCase(scheme)) {
            port = 443;
        }
    }

    if (!HttpScheme.HTTP.equalsIgnoreCase(scheme) && !HttpScheme.HTTPS.equalsIgnoreCase(scheme)) {
        if (log.isErrorEnabled()) {
            log.error("仅有HTTP（S）是支持的。");
        }
        return;
    }

    final boolean ssl = HttpScheme.HTTPS.equalsIgnoreCase(scheme);

    this.setSSlContext(ssl);

    final Bootstrap b = new Bootstrap();
    b.group(GROUP)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .remoteAddress(InetSocketAddress.createUnresolved(host, port))
    ;

    channelPool = new FixedChannelPool(b, new HttpClientChannelPoolHandler(sslContext), maxCount);
}
 

FastdfsPool(Bootstrap bootstrap, long readTimeout, long idleTimeout, int maxConnPerHost, int maxPendingRequests) {
    this.channelPool = new FixedChannelPool(
            bootstrap,
            new FastdfsPoolHandler(readTimeout, idleTimeout),
            maxConnPerHost,
            maxPendingRequests
    );
}
 

@Override
protected FixedChannelPool newPool(InetSocketAddress key) {
	return new FixedChannelPool(initBootStrap().remoteAddress(key), new InitChannelPoolHandler(), new DefaultChannelHealthChecker(),
               AcquireTimeoutAction.NEW, 500, maxConnections, maxPending);
}
 

FastdfsPool(Bootstrap bootstrap, long readTimeout, long idleTimeout, int maxConnPerHost) {
    this.channelPool = new FixedChannelPool(bootstrap,
        new FastdfsPoolHandler(readTimeout, idleTimeout), maxConnPerHost);
}