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

public static void main(String[] args) {
        EventLoopGroup group = new NioEventLoopGroup();
        final Bootstrap cb = new Bootstrap();
        cb.group(group).channel(NioSocketChannel.class);
        InetSocketAddress addr1 = new InetSocketAddress("10.0.0.10", 8888);
        InetSocketAddress addr2 = new InetSocketAddress("10.0.0.11", 8888);

        //连接池map
        ChannelPoolMap<InetSocketAddress, SimpleChannelPool> poolMap = new AbstractChannelPoolMap<InetSocketAddress, SimpleChannelPool>() {
            @Override
            protected SimpleChannelPool newPool(InetSocketAddress key) {
                return new SimpleChannelPool(cb.remoteAddress(key), new TestChannelPoolHandler());
            }
        };

        final SimpleChannelPool pool1 = poolMap.get(addr1);//取出連接addr1地址的连接池
        final SimpleChannelPool pool2 = poolMap.get(addr2);//取出連接addr2地址的连接池
        Future<Channel> f1 = pool1.acquire();//获取一个连接
        f1.addListener(new FutureListener<Channel>() {
            @Override
            public void operationComplete(Future<Channel> f) {
                if (f.isSuccess()) {
                    Channel ch = f.getNow();
                   //连接地址1的某个channel
                    //使用连接发送消息
//                  ch.write(msg)
                    //用完释放
                    pool1.release(ch);
                }
            }
        });

    }

public XioConnectionPool(Bootstrap bootstrap, AsyncRetryLoopFactory retryLoopFactory) {
  Preconditions.checkNotNull(bootstrap);
  this.retryLoopFactory = Preconditions.checkNotNull(retryLoopFactory);
  handler = bootstrap.config().handler();
  eventLoopGroup = bootstrap.config().group();
  simpleChannelPool = new SimpleChannelPool(bootstrap, channelPoolHandler, channelHealthChecker);
}

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

@Override
public ChannelPool createChannelPool(InetSocketAddress key) {
    return new SimpleChannelPool(getBootstrap().remoteAddress(key), channelPoolHandler);
}

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