Java 类org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor 实例源码

private CloseableHttpAsyncClient createHttpAsyncClient(YunpianConf conf) throws IOReactorException {
    IOReactorConfig ioReactorConfig = IOReactorConfig.custom().setIoThreadCount(Runtime.getRuntime().availableProcessors())
            .setConnectTimeout(conf.getConfInt(YunpianConf.HTTP_CONN_TIMEOUT, "10000"))
            .setSoTimeout(conf.getConfInt(YunpianConf.HTTP_SO_TIMEOUT, "30000")).build();
    ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);

    PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(ioReactor);
    ConnectionConfig connectionConfig = ConnectionConfig.custom().setMalformedInputAction(CodingErrorAction.IGNORE)
            .setUnmappableInputAction(CodingErrorAction.IGNORE)
            .setCharset(Charset.forName(conf.getConf(YunpianConf.HTTP_CHARSET, YunpianConf.HTTP_CHARSET_DEFAULT))).build();
    connManager.setDefaultConnectionConfig(connectionConfig);
    connManager.setMaxTotal(conf.getConfInt(YunpianConf.HTTP_CONN_MAXTOTAL, "100"));
    connManager.setDefaultMaxPerRoute(conf.getConfInt(YunpianConf.HTTP_CONN_MAXPERROUTE, "10"));

    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setConnectionManager(connManager).build();
    httpclient.start();
    return httpclient;
}

public FiberApacheHttpClientRequestExecutor(final Validator<CloseableHttpResponse> resValidator, final int maxConnections, final int timeout, final int parallelism) throws IOReactorException {
  final DefaultConnectingIOReactor ioreactor = new DefaultConnectingIOReactor(IOReactorConfig.custom().
    setConnectTimeout(timeout).
    setIoThreadCount(parallelism).
    setSoTimeout(timeout).
    build());

  final PoolingNHttpClientConnectionManager mngr = new PoolingNHttpClientConnectionManager(ioreactor);
  mngr.setDefaultMaxPerRoute(maxConnections);
  mngr.setMaxTotal(maxConnections);

  final CloseableHttpAsyncClient ahc = HttpAsyncClientBuilder.create().
    setConnectionManager(mngr).
    setDefaultRequestConfig(RequestConfig.custom().setLocalAddress(null).build()).build();

  client = new FiberHttpClient(ahc);
  validator = resValidator;
}

public AsyncConnectionManager(
        HttpHost target,
        int maxConnections,
        NHttpClientHandler handler, 
        HttpParams params) throws IOReactorException {
    super();
    this.target = target;
    this.maxConnections = maxConnections;
    this.handler = handler;
    this.params = params;
    this.lock = new Object();
    this.allConns = new HashSet<NHttpClientConnection>();
    this.availableConns = new LinkedList<NHttpClientConnection>();
    this.pendingRequests = new LinkedList<AsyncConnectionRequest>();
    this.ioreactor = new DefaultConnectingIOReactor(2, params);
}

private DashboardSetupStatus initDashboard(final String hostAddress, final int port) {
  final String dashboardURL = String.format("http://%s:%d/", hostAddress, port);
  try {
    // Create a pool of http client connection, which allow up to Integer.MAX_VALUE connections.
    final PoolingNHttpClientConnectionManager connectionManager
        = new PoolingNHttpClientConnectionManager(new DefaultConnectingIOReactor());
    connectionManager.setMaxTotal(Integer.MAX_VALUE);
    final CloseableHttpAsyncClient reusableHttpClient =
        HttpAsyncClients.custom().setConnectionManager(connectionManager).build();
    reusableHttpClient.start();

    // run another thread to send metrics.
    runMetricsSenderThread();

    return DashboardSetupStatus.getSuccessful(dashboardURL, reusableHttpClient);
  } catch (IOReactorException e) {
    LOG.log(Level.WARNING, "Dashboard: Fail on initializing connection to the dashboard server.", e);
    return DashboardSetupStatus.getFailed();
  }
}

/**
 * Used internally to initialize the internal HTTP client used by all
 * instances of a client.
 * <p>
 * This method can be overriden to provide a client with different options.
 * The client built gets an extra interceptor to add the credentials headers.
 *
 * @return HTTP default async client builder
 */
protected static HttpAsyncClientBuilder defaultClientBuilder() {

  try {
    DefaultConnectingIOReactor ioReactor = new DefaultConnectingIOReactor();
    connMgr = new PoolingNHttpClientConnectionManager(ioReactor);
    connMgr.setMaxTotal(maxConnections);
    connMgr.setDefaultMaxPerRoute(maxConnections);
  } catch (IOReactorException e) {
  }

  return HttpAsyncClients
      .custom()
      .addInterceptorLast(new GzipInterceptors.GzipRequestInterceptor())
      .setConnectionManager(connMgr)
      .setDefaultRequestConfig(
          RequestConfig.custom()
          .setSocketTimeout(3600 * 1000) // 1 hour
              .build())
      .setKeepAliveStrategy(keepAliveStrategy);
}

@Bean
public CloseableHttpAsyncClient asyncHttpClient() {
    try {
        PoolingNHttpClientConnectionManager connectionManager = new PoolingNHttpClientConnectionManager(
                new DefaultConnectingIOReactor(IOReactorConfig.DEFAULT));
        connectionManager.setMaxTotal(DEFAULT_MAX_TOTAL_CONNECTIONS);
        connectionManager.setDefaultMaxPerRoute(DEFAULT_MAX_CONNECTIONS_PER_ROUTE);
        RequestConfig config = RequestConfig.custom()
                .setConnectTimeout(DEFAULT_READ_TIMEOUT_MILLISECONDS)
                .build();

        return HttpAsyncClientBuilder.create()
                .setConnectionManager(connectionManager)
                .setDefaultRequestConfig(config)
                .build();
    } catch (Exception e) {
        throw Throwables.propagate(e);
    }
}

public SleepServerApiClient() throws Exception {
    connectionManager = new PoolingNHttpClientConnectionManager(
            new DefaultConnectingIOReactor(IOReactorConfig.DEFAULT));
    connectionManager.setMaxTotal(20000);
    connectionManager.setDefaultMaxPerRoute(20000);

    RequestConfig config = RequestConfig.custom().setConnectTimeout(120000)
            .build();

    CloseableHttpAsyncClient httpClient = HttpAsyncClientBuilder.create()
            .setConnectionManager(connectionManager)
            .setDefaultRequestConfig(config).build();

    HttpComponentsAsyncClientHttpRequestFactory requestFactory = new HttpComponentsAsyncClientHttpRequestFactory(
            httpClient);
    client = new AsyncRestTemplate(requestFactory);
}

private NHttpClientConnectionManager getNHttpConnManager(Config config) throws IOException {
  NHttpClientConnectionManager httpConnManager;

  String connMgrStr = config.getString(HTTP_CONN_MANAGER);
  switch (ApacheHttpClient.ConnManager.valueOf(connMgrStr.toUpperCase())) {
    case POOLING:
      ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor();
      PoolingNHttpClientConnectionManager poolingConnMgr = new PoolingNHttpClientConnectionManager(ioReactor);
      poolingConnMgr.setMaxTotal(config.getInt(POOLING_CONN_MANAGER_MAX_TOTAL_CONN));
      poolingConnMgr.setDefaultMaxPerRoute(config.getInt(POOLING_CONN_MANAGER_MAX_PER_CONN));
      httpConnManager = poolingConnMgr;
      break;
    default:
      throw new IllegalArgumentException(connMgrStr + " is not supported");
  }

  LOG.info("Using " + httpConnManager.getClass().getSimpleName());
  return httpConnManager;
}

public AsyncServiceClient(ClientConfiguration config) {
    super(config);
    try {
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setIoThreadCount(config.getIoThreadCount()).build();
        ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(
                ioReactorConfig);
        PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(
                ioReactor);
        cm.setMaxTotal(config.getMaxConnections());
        cm.setDefaultMaxPerRoute(config.getMaxConnections());
        httpClient = new HttpFactory().createHttpAsyncClient(config, cm);

        /*
         * socketTimeout的值限制了closeIdleConnections执行的周期。
         * 如果周期相对socketTimeout的值过长，有可能一个请求分配到一个即将socketTimeout的连接，
         * 在请求发送之前即抛出SocketTimeoutException。
         * 现在让closeIdleConnections的执行周期为socketTimeout / 2.5。
         */
        long closePeriod = 5000;
        if (config.getSocketTimeoutInMillisecond() > 0) {
            closePeriod = (long) (config.getSocketTimeoutInMillisecond() / 2.5);
        }
        closePeriod = closePeriod < 5000 ? closePeriod : 5000;
        connEvictor = new IdleConnectionEvictor(cm, closePeriod);
        httpClient.start();
        connEvictor.start();
    } catch (IOReactorException ex) {
        throw new ClientException(String.format("IOReactorError: %s",
                ex.getMessage()), ex);
    }
}

@Bean
PoolingNHttpClientConnectionManager poolingNHttpClientConnectionManager() throws IOReactorException {
    PoolingNHttpClientConnectionManager connectionManager = new PoolingNHttpClientConnectionManager(
            new DefaultConnectingIOReactor(IOReactorConfig.DEFAULT));
    connectionManager.setMaxTotal(maxTotalConnections);
    connectionManager.setDefaultMaxPerRoute(maxConnectionsPerRoute);
    return connectionManager;
}

public RmendRequestAdapter(String _targetHost, String _endpointTemplate) throws IOReactorException {
    ioReactor = new DefaultConnectingIOReactor();
    cm = new PoolingNHttpClientConnectionManager(ioReactor);
    cm.setMaxTotal(20);

    httpclient = HttpAsyncClients.createPipelining();
    this.targetHost = _targetHost;
    this.endpointTemplate = _endpointTemplate;
}

@Bean
public CloseableHttpAsyncClient asyncHttpClient() throws Exception {
    PoolingNHttpClientConnectionManager connectionManager = new PoolingNHttpClientConnectionManager(new DefaultConnectingIOReactor(
            IOReactorConfig.DEFAULT));
    connectionManager.setMaxTotal(DEFAULT_MAX_TOTAL_CONNECTIONS);
    connectionManager.setDefaultMaxPerRoute(DEFAULT_MAX_CONNECTIONS_PER_ROUTE);
    connectionManager.setMaxPerRoute(new HttpRoute(new HttpHost("localhost")), 20);
    RequestConfig config = RequestConfig.custom().setConnectTimeout(DEFAULT_READ_TIMEOUT_MILLISECONDS).build();

    CloseableHttpAsyncClient httpclient = HttpAsyncClientBuilder.create().setConnectionManager(connectionManager).setDefaultRequestConfig(config)
            .build();
    return httpclient;
}

private HttpAsyncClientBuilder initialize() {
    try {
        final PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(
            new DefaultConnectingIOReactor( IOReactorConfig.custom()
                .setConnectTimeout( connectTimeout )
                .setSoTimeout( readTimeout )
                .build() ),
            RegistryBuilder.<SchemeIOSessionStrategy>create()
                .register( "http", NoopIOSessionStrategy.INSTANCE )
                .register( "https",
                    new SSLIOSessionStrategy( certificateLocation != null ?
                        createSSLContext( certificateLocation, certificatePassword )
                        : SSLContexts.createDefault(),
                        split( System.getProperty( "https.protocols" ) ),
                        split( System.getProperty( "https.cipherSuites" ) ),
                        new DefaultHostnameVerifier( PublicSuffixMatcherLoader.getDefault() ) ) )
                .build() );

        connManager.setMaxTotal( maxConnTotal );
        connManager.setDefaultMaxPerRoute( maxConnPerRoute );

        return ( certificateLocation != null ?
            HttpAsyncClients.custom()
                .setSSLContext( createSSLContext( certificateLocation, certificatePassword ) )
            : HttpAsyncClients.custom() )
            .setMaxConnPerRoute( maxConnPerRoute )
            .setConnectionManager( connManager )
            .setMaxConnTotal( maxConnTotal )
            .setKeepAliveStrategy( DefaultConnectionKeepAliveStrategy.INSTANCE )
            .setDefaultRequestConfig( RequestConfig
                .custom()
                .setRedirectsEnabled( redirectsEnabled )
                .setCookieSpec( cookieSpec )
                .build() )
            .setDefaultCookieStore( basicCookieStore );
    } catch( IOReactorException e ) {
        throw new UncheckedIOException( e );
    }
}

/**
 * Create connection manager for asynchronous http client.
 *
 * @return Connection manager for asynchronous http client.
 * @throws IOReactorException in case if a non-recoverable I/O error.
 */
protected NHttpClientConnectionManager createNHttpClientConnectionManager() throws IOReactorException {
    ConnectingIOReactor ioReactor =
            new DefaultConnectingIOReactor(IOReactorConfig.custom()
                    .setSoTimeout(this.config.getSocketTimeoutInMillis()).setTcpNoDelay(true).build());
    PoolingNHttpClientConnectionManager connectionManager = new PoolingNHttpClientConnectionManager(ioReactor);
    connectionManager.setDefaultMaxPerRoute(this.config.getMaxConnections());
    connectionManager.setMaxTotal(this.config.getMaxConnections());
    return connectionManager;
}

public AsyncClient() throws IOException {

        // Create I/O reactor configuration
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setIoThreadCount(Runtime.getRuntime().availableProcessors())
                .setTcpNoDelay(true)
                .build();

        // Create a custom I/O reactor
        ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);

        // Create a custom Connection Manager
        connManager = new PoolingNHttpClientConnectionManager(ioReactor);
        connManager.setDefaultMaxPerRoute(defaultMaxPerRoute);
        connManager.setMaxTotal(defaultMaxPerRoute * MAX_CONN_MULTIPLICATION);

        // Create global request configuration
        RequestConfig defaultRequestConfig = RequestConfig.custom()
                .setCookieSpec(CookieSpecs.DEFAULT)
                .setExpectContinueEnabled(true)
                .setTargetPreferredAuthSchemes(Arrays.asList(AuthSchemes.NTLM, AuthSchemes.DIGEST))
                .setProxyPreferredAuthSchemes(Arrays.asList(AuthSchemes.BASIC))
                .build();

        // Update the request level configuration
        updateRequestConfig();

        // Create a custom Redirect Handler
        redirectStrategy = new RedirectHandler(requestConfig);

        httpAsyncClient = HttpAsyncClients.custom()
                .setConnectionManager(connManager)
                .setDefaultRequestConfig(defaultRequestConfig)
                .setRedirectStrategy(redirectStrategy)
                .build();
    }

protected ExecCallbackAsyncREST<HttpResponse> buildAsyncClient(RESTPool pool) throws IOException {
    SSLContext sslContext;
    try {
        sslContext = SSLContext.getDefault();
    } catch (NoSuchAlgorithmException e) {
        throw new IOException(e);
    }

    Registry<SchemeIOSessionStrategy> socketRegistry = RegistryBuilder.<SchemeIOSessionStrategy>create()
            .register("http", NoopIOSessionStrategy.INSTANCE)
            .register("https", new SSLIOSessionStrategy(sslContext, NoopHostnameVerifier.INSTANCE))
            .build();

    IOReactorConfig socketConfig = IOReactorConfig.custom()
            .setIoThreadCount(pool.getReactorThreadCount())
            .setSoTimeout(new Long(pool.getSocketTimeout()).intValue())
            .setTcpNoDelay(true)
            .setSoKeepAlive(true)
            .setSelectInterval(REACTOR_SELECT_INTERVAL)
            .build();

    ConnectionConfig connectionConfig = ConnectionConfig.custom()
            .setCharset(StandardCharsets.UTF_8)
            .setMalformedInputAction(CodingErrorAction.IGNORE)
            .setUnmappableInputAction(CodingErrorAction.IGNORE)
            .build();

    RequestConfig requestConfig = RequestConfig.custom()
            .setConnectionRequestTimeout(new Long(pool.getMaxPoolWait()).intValue())
            .setConnectTimeout(new Long(pool.getConnectionTimeout()).intValue())
            .setExpectContinueEnabled(pool.expectContinue())
            .setRedirectsEnabled(false)
            .setStaleConnectionCheckEnabled(pool.getValidationOnInactivity() >= 0)
            .build();

    NHttpConnectionFactory<ManagedNHttpClientConnection> connFactory = new ManagedNHttpClientConnectionFactory(
            new org.apache.http.impl.nio.codecs.DefaultHttpRequestWriterFactory(),
            new org.apache.http.impl.nio.codecs.DefaultHttpResponseParserFactory(),
            HeapByteBufferAllocator.INSTANCE
    );

    //TODO set validateAfterInactivity when supported
    PoolingNHttpClientConnectionManager ccm = new PoolingNHttpClientConnectionManager(
            new DefaultConnectingIOReactor(socketConfig),
            connFactory,
            socketRegistry,
            new SystemDefaultDnsResolver()
    );

    ccm.setMaxTotal(pool.getMaxTotal());
    ccm.setDefaultMaxPerRoute(pool.getMaxPerRoute());
    ccm.setDefaultConnectionConfig(connectionConfig);

    HttpAsyncClientBuilder builder = HttpAsyncClients.custom()
            .setConnectionManager(ccm)
            .setDefaultRequestConfig(requestConfig)
            .setKeepAliveStrategy(DefaultConnectionKeepAliveStrategy.INSTANCE)
            .disableCookieManagement();

    IdleAsyncConnectionEvictor evictor = new IdleAsyncConnectionEvictor(ccm, pool.getEvictorSleep(), TimeUnit.MILLISECONDS, pool.getMaxIdleTime(), TimeUnit.MILLISECONDS);

    addProxy(pool, builder);

    handleRedirects(pool, builder);

    CloseableHttpAsyncClient servClient = builder.build();

    servClient.start();

    HTTPCClientMonitor monitor = pool.hasConnectionMetrics() ? new HTTPCAsyncClientMonitor(pool.getName(), ccm) : null;

    return new HTTPCAsyncClient(servClient, evictor, monitor);
}

public RestClient build() {
        try {
            // Create common default configuration
            RequestConfig clientConfig = RequestConfig.custom()
                    .setRedirectsEnabled(followredirect)
                    .setConnectTimeout(connectionTimeout)
                    .setSocketTimeout(socketTimeout)
                    .setConnectionRequestTimeout(socketTimeout)
                    .setProxy(proxy)
                    .setCookieSpec(cookieSpec)
                    .build();

            PoolingHttpClientConnectionManager syncConnectionManager = new PoolingHttpClientConnectionManager();
            syncConnectionManager.setMaxTotal(maxTotal);
//            syncConnectionManager.setDefaultMaxPerRoute(maxPerRoute);


            CloseableHttpClient syncClient = HttpClientBuilder.create()
                    .setDefaultRequestConfig(clientConfig)
                    .setConnectionManager(syncConnectionManager)
                    .build();


            DefaultConnectingIOReactor ioreactor = new DefaultConnectingIOReactor();
            PoolingNHttpClientConnectionManager asyncConnectionManager = new PoolingNHttpClientConnectionManager(ioreactor);
            asyncConnectionManager.setMaxTotal(maxTotal);
//                asyncConnectionManager.setDefaultMaxPerRoute(maxPerRoute);

            CloseableHttpAsyncClient asyncClient = HttpAsyncClientBuilder.create()
                    .setDefaultRequestConfig(clientConfig)
                    .setConnectionManager(asyncConnectionManager)
                    .build();


            RestClient restClient = new RestClient(baseUrl, objectMapper, defaultHeaders, urlTransformer, asyncConnectionManager, syncConnectionManager, asyncClient, syncClient);
            ClientContainer.addClient(restClient);
            return restClient;

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

public static void main(String[] args) throws Exception {
    ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor();
    PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(ioReactor);
    cm.setMaxTotal(100);
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setConnectionManager(cm).build();
    try {
        httpclient.start();

        // create an array of URIs to perform GETs on
        String[] urisToGet = { "http://hc.apache.org/", "http://hc.apache.org/httpcomponents-core-ga/",
                "http://hc.apache.org/httpcomponents-client-ga/", };

        IdleConnectionEvictor connEvictor = new IdleConnectionEvictor(cm);
        connEvictor.start();

        final CountDownLatch latch = new CountDownLatch(urisToGet.length);
        for (final String uri : urisToGet) {
            final HttpGet httpget = new HttpGet(uri);
            httpclient.execute(httpget, new FutureCallback<HttpResponse>() {

                @Override
                public void completed(final HttpResponse response) {
                    latch.countDown();
                    System.out.println(httpget.getRequestLine() + "->" + response.getStatusLine());
                }

                @Override
                public void failed(final Exception ex) {
                    latch.countDown();
                    System.out.println(httpget.getRequestLine() + "->" + ex);
                }

                @Override
                public void cancelled() {
                    latch.countDown();
                    System.out.println(httpget.getRequestLine() + " cancelled");
                }

            });
        }
        latch.await();

        // Sleep 10 sec and let the connection evictor do its job
        Thread.sleep(20000);

        // Shut down the evictor thread
        connEvictor.shutdown();
        connEvictor.join();

    } finally {
        httpclient.close();
    }
}

public static void main(String[] args) throws Exception {
    HttpParams params = new SyncBasicHttpParams();
    params
        .setIntParameter(CoreConnectionPNames.SO_TIMEOUT, 5000)
        .setIntParameter(CoreConnectionPNames.CONNECTION_TIMEOUT, 10000)
        .setIntParameter(CoreConnectionPNames.SOCKET_BUFFER_SIZE, 8 * 1024)
        .setBooleanParameter(CoreConnectionPNames.STALE_CONNECTION_CHECK, false)
        .setBooleanParameter(CoreConnectionPNames.TCP_NODELAY, true)
        .setParameter(CoreProtocolPNames.USER_AGENT, "HttpComponents/1.1");

    final ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(2, params);

    HttpProcessor httpproc = new ImmutableHttpProcessor(new HttpRequestInterceptor[] {
            new RequestContent(),
            new RequestTargetHost(),
            new RequestConnControl(),
            new RequestUserAgent(),
            new RequestExpectContinue()});

    // We are going to use this object to synchronize between the 
    // I/O event and main threads
    CountDownLatch requestCount = new CountDownLatch(3);

    BufferingHttpClientHandler handler = new BufferingHttpClientHandler(
            httpproc,
            new MyHttpRequestExecutionHandler(requestCount),
            new DefaultConnectionReuseStrategy(),
            params);

    handler.setEventListener(new EventLogger());

    final IOEventDispatch ioEventDispatch = new DefaultClientIOEventDispatch(handler, params);

    Thread t = new Thread(new Runnable() {

        public void run() {
            try {
                ioReactor.execute(ioEventDispatch);
            } catch (InterruptedIOException ex) {
                System.err.println("Interrupted");
            } catch (IOException e) {
                System.err.println("I/O error: " + e.getMessage());
            }
            System.out.println("Shutdown");
        }

    });
    t.start();

    SessionRequest[] reqs = new SessionRequest[3];
    reqs[0] = ioReactor.connect(
            new InetSocketAddress("www.yahoo.com", 80), 
            null, 
            new HttpHost("www.yahoo.com"),
            new MySessionRequestCallback(requestCount));
    reqs[1] = ioReactor.connect(
            new InetSocketAddress("www.google.com", 80), 
            null,
            new HttpHost("www.google.ch"),
            new MySessionRequestCallback(requestCount));
    reqs[2] = ioReactor.connect(
            new InetSocketAddress("www.apache.org", 80), 
            null,
            new HttpHost("www.apache.org"),
            new MySessionRequestCallback(requestCount));

    // Block until all connections signal
    // completion of the request execution
    requestCount.await();

    System.out.println("Shutting down I/O reactor");

    ioReactor.shutdown();

    System.out.println("Done");
}

public static void main(String[] args) throws Exception {
    ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor();
    PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(ioReactor);
    cm.setMaxTotal(100);
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom()
            .setConnectionManager(cm)
            .build();
    try {
        httpclient.start();

        // create an array of URIs to perform GETs on
        String[] urisToGet = {
            "http://hc.apache.org/",
            "http://hc.apache.org/httpcomponents-core-ga/",
            "http://hc.apache.org/httpcomponents-client-ga/",
        };

        IdleConnectionEvictor connEvictor = new IdleConnectionEvictor(cm);
        connEvictor.start();

        final CountDownLatch latch = new CountDownLatch(urisToGet.length);
        for (final String uri: urisToGet) {
            final HttpGet httpget = new HttpGet(uri);
            httpclient.execute(httpget, new FutureCallback<HttpResponse>() {

                @Override
                public void completed(final HttpResponse response) {
                    latch.countDown();
                    System.out.println(httpget.getRequestLine() + "->" + response.getStatusLine());
                }

                @Override
                public void failed(final Exception ex) {
                    latch.countDown();
                    System.out.println(httpget.getRequestLine() + "->" + ex);
                }

                @Override
                public void cancelled() {
                    latch.countDown();
                    System.out.println(httpget.getRequestLine() + " cancelled");
                }

            });
        }
        latch.await();

        // Sleep 10 sec and let the connection evictor do its job
        Thread.sleep(20000);

        // Shut down the evictor thread
        connEvictor.shutdown();
        connEvictor.join();

    } finally {
        httpclient.close();
    }
}