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

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 static HttpClient createHttpClient(HiTSDBConfig config) throws HttpClientInitException {
    Objects.requireNonNull(config);

    // 创建 ConnectingIOReactor
    ConnectingIOReactor ioReactor = initIOReactorConfig(config);

    // 创建链接管理器
    final PoolingNHttpClientConnectionManager cm = new PoolingNHttpClientConnectionManager(ioReactor);

    // 创建令牌管理器
    semaphoreManager = createSemaphoreManager(config);

    // 创建HttpAsyncClient
    CloseableHttpAsyncClient httpAsyncClient = createPoolingHttpClient(config,cm,semaphoreManager);

    // 组合生产HttpClientImpl
    HttpClient httpClientImpl = new HttpClient(config,httpAsyncClient,semaphoreManager);

    return httpClientImpl;
}

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 ListeningHandler(
        final HttpHost targetHost,
        final ConnectingIOReactor connectingIOReactor,
        final HttpProcessor httpProcessor, 
        final HttpResponseFactory responseFactory,
        final ConnectionReuseStrategy connStrategy,
        final HttpParams params) {
    super();
    this.targetHost = targetHost;
    this.connectingIOReactor = connectingIOReactor;
    this.httpProcessor = httpProcessor;
    this.connStrategy = connStrategy;
    this.responseFactory = responseFactory;
    this.params = params;
}

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

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

private void initialize() {
    if (initialized.getAndSet(true)) {
        return;
    }
    IOReactorConfig.Builder config = createConfig();
    // params.setParameter(CoreProtocolPNames.USER_AGENT, "jsonrpc4j/1.0");
    final ConnectingIOReactor ioReactor = createIoReactor(config);
    createSslContext();
    int socketBufferSize = Integer.getInteger("com.googlecode.jsonrpc4j.async.socket.buffer", 8 * 1024);
    final ConnectionConfig connectionConfig = ConnectionConfig.custom().setBufferSize(socketBufferSize).build();
    BasicNIOConnFactory nioConnFactory = new BasicNIOConnFactory(sslContext, null, connectionConfig);
    pool = new BasicNIOConnPool(ioReactor, nioConnFactory, Integer.getInteger("com.googlecode.jsonrpc4j.async.connect.timeout", 30000));
    pool.setDefaultMaxPerRoute(Integer.getInteger("com.googlecode.jsonrpc4j.async.max.inflight.route", 500));
    pool.setMaxTotal(Integer.getInteger("com.googlecode.jsonrpc4j.async.max.inflight.total", 500));

    Thread t = new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                HttpAsyncRequestExecutor protocolHandler = new HttpAsyncRequestExecutor();
                IOEventDispatch ioEventDispatch = new DefaultHttpClientIODispatch(protocolHandler, sslContext, connectionConfig);
                ioReactor.execute(ioEventDispatch);
            } catch (InterruptedIOException ex) {
                System.err.println("Interrupted");
            } catch (IOException e) {
                System.err.println("I/O error: " + e.getMessage());
            }
        }
    }, "jsonrpc4j HTTP IOReactor");

    t.setDaemon(true);
    t.start();

    HttpProcessor httpProcessor = new ImmutableHttpProcessor(new RequestContent(), new RequestTargetHost(), new RequestConnControl(), new RequestUserAgent(), new RequestExpectContinue(false));
    requester = new HttpAsyncRequester(httpProcessor, new DefaultConnectionReuseStrategy());
}

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

public ProxyConnPool(
        final ConnectingIOReactor ioreactor,
        final ConnectionConfig config) {
    super(ioreactor, config);
}

public ProxyConnPool(
        final ConnectingIOReactor ioreactor,
        final NIOConnFactory<HttpHost, NHttpClientConnection> connFactory,
        final int connectTimeout) {
    super(ioreactor, connFactory, connectTimeout);
}