Java 类io.netty.channel.pool.ChannelPoolHandler 实例源码

public BasicMultiNodeConnPoolMock(
    final Semaphore concurrencyThrottle, final String[] nodes, final Bootstrap bootstrap,
    final ChannelPoolHandler connPoolHandler, final int defaultPort,
    final int connFailSeqLenLimit
) {
    super(
        concurrencyThrottle, nodes, bootstrap, connPoolHandler, defaultPort,
        connFailSeqLenLimit
    );
}

private HubManager(final Properties properties) {
        Runtime.getRuntime().addShutdownHook(new Thread(){
            public void run() { try { close(); } catch (Exception x) {/* No Op */} }
        });
        log.info(">>>>> Initializing HubManager...");
        metricMetaService = new MetricsMetaAPIImpl(properties);
        tsdbEndpoint = TSDBEndpoint.getEndpoint(metricMetaService.getSqlWorker());
        for(String url: tsdbEndpoint.getUpServers()) {
            final URL tsdbUrl = URLHelper.toURL(url);
            tsdbAddresses.add(new InetSocketAddress(tsdbUrl.getHost(), tsdbUrl.getPort()));
        }
        endpointCount = tsdbAddresses.size(); 
        endpointSequence = new AtomicInteger(endpointCount);
        group = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2, metricMetaService.getForkJoinPool());
        bootstrap = new Bootstrap();
        bootstrap
            .handler(channelInitializer)
            .group(group)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator())
            .option(ChannelOption.ALLOCATOR, BufferManager.getInstance());
        final ChannelPoolHandler poolHandler = this;
        poolMap = new AbstractChannelPoolMap<InetSocketAddress, SimpleChannelPool>() {
            @Override
            protected SimpleChannelPool newPool(final InetSocketAddress key) {
                final Bootstrap b = new Bootstrap().handler(channelInitializer)
                .group(group)
                .remoteAddress(key)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator())
                .option(ChannelOption.ALLOCATOR, BufferManager.getInstance());              
                return new SimpleChannelPool(b, poolHandler);
            }
        };
        eventExecutor = new DefaultEventExecutor(metricMetaService.getForkJoinPool());
        channelGroup = new DefaultChannelGroup("MetricHubChannelGroup", eventExecutor);

//      tsdbAddresses.parallelStream().forEach(addr -> {
//          final Set<Channel> channels = Collections.synchronizedSet(new HashSet<Channel>(3));
//          IntStream.of(1,2,3).parallel().forEach(i -> {
//              final ChannelPool pool = poolMap.get(addr); 
//              try {channels.add(pool.acquire().awaitUninterruptibly().get());
//              } catch (Exception e) {}
//              log.info("Acquired [{}] Channels", channels.size());
//              channels.parallelStream().forEach(ch -> pool.release(ch));
//          });
//      });




        log.info("<<<<< HubManager Initialized.");
    }

public ConnectionlessChannelPool(Bootstrap bootstrap, ChannelPoolHandler handler) {
    super(bootstrap, handler);
}

public ConnectionlessChannelPool(Bootstrap bootstrap, ChannelPoolHandler handler, ChannelHealthChecker healthCheck) {
    super(bootstrap, handler, healthCheck);
}

public ConnectionlessChannelPool(Bootstrap bootstrap, ChannelPoolHandler handler, ChannelHealthChecker healthCheck, boolean releaseHealthCheck) {
    super(bootstrap, handler, healthCheck, releaseHealthCheck);
}

/**
 * @return The {@link ChannelPoolHandler}
 */
public ChannelPoolHandler getChannelPoolHandler() {
    return channelPoolHandler;
}

ChannelPool newPool(Bootstrap b,
ChannelPoolHandler handler,
ChannelHealthChecker checker);