Java 类org.apache.http.impl.nio.client.HttpAsyncClients 实例源码

@Override
public CloseableHttpAsyncClient generateClient ()
{
   CredentialsProvider credsProvider = new BasicCredentialsProvider();
   credsProvider.setCredentials(new AuthScope (AuthScope.ANY),
           new UsernamePasswordCredentials(serviceUser, servicePass));
   RequestConfig rqconf = RequestConfig.custom()
         .setCookieSpec(CookieSpecs.DEFAULT)
         .setSocketTimeout(Timeouts.SOCKET_TIMEOUT)
         .setConnectTimeout(Timeouts.CONNECTION_TIMEOUT)
         .setConnectionRequestTimeout(Timeouts.CONNECTION_REQUEST_TIMEOUT)
         .build();
   CloseableHttpAsyncClient res = HttpAsyncClients.custom ()
         .setDefaultCredentialsProvider (credsProvider)
         .setDefaultRequestConfig(rqconf)
         .build ();
   res.start ();
   return res;
}

public void send(List<Map<Object, Object>> events, AsyncSuccessCallback<ProducedEventsResult> onSuccess, AsyncFailCallback onFail, AsyncCancelledCallback onCancel) throws IOException, InterruptedException {
    CloseableHttpAsyncClient httpClient = HttpAsyncClients.createDefault();
    httpClient.start();

    String url = String.format("%s/%s/bulk-produce", this.endpoint, this.topicId);
    System.out.println(url);
    HttpPost httpPost = new HttpPost(url);
    httpPost.addHeader("Authorization", this.writeApiKey);
    httpPost.addHeader("Content-type", this.format);

    String jsonString = MAPPER.writeValueAsString(events);
    HttpEntity entity = new ByteArrayEntity(jsonString.getBytes());
    httpPost.setEntity(entity);

    ResponseParser<ProducedEventsResult> parser = new BulkProduceEventsParser();
    AsyncCallback cb = new AsyncCallback(httpClient, parser, MAPPER, onSuccess, onFail, onCancel);
    httpClient.execute(httpPost, cb);
}

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

@Override
public CloseableHttpAsyncClient generateClient ()
{
   CredentialsProvider credsProvider = new BasicCredentialsProvider();
   credsProvider.setCredentials(new AuthScope (AuthScope.ANY),
            new UsernamePasswordCredentials(username, password));
   RequestConfig rqconf = RequestConfig.custom()
         .setCookieSpec(CookieSpecs.DEFAULT)
         .setSocketTimeout(Timeouts.SOCKET_TIMEOUT)
         .setConnectTimeout(Timeouts.CONNECTION_TIMEOUT)
         .setConnectionRequestTimeout(Timeouts.CONNECTION_REQUEST_TIMEOUT)
         .build();
   CloseableHttpAsyncClient res = HttpAsyncClients.custom ()
         .setDefaultCredentialsProvider (credsProvider)
         .setDefaultRequestConfig(rqconf)
         .build ();
   res.start ();
   return res;
}

public void send(Map<Object, Object> event, AsyncSuccessCallback<ProducedEventResult> onSuccess, AsyncFailCallback onFail, AsyncCancelledCallback onCancel) throws IOException {
    CloseableHttpAsyncClient httpClient = HttpAsyncClients.createDefault();
    httpClient.start();

    String url = String.format("%s/%s/produce", this.endpoint, this.topicId);
    HttpPost httpPost = new HttpPost(url);
    httpPost.addHeader("Authorization", this.writeApiKey);
    httpPost.addHeader("Content-type", this.format);

    String jsonString = MAPPER.writeValueAsString(event);
    HttpEntity entity = new ByteArrayEntity(jsonString.getBytes());
    httpPost.setEntity(entity);

    ResponseParser<ProducedEventResult> parser = new ProduceEventParser();
    AsyncCallback cb = new AsyncCallback(httpClient, parser, MAPPER, onSuccess, onFail, onCancel);
    httpClient.execute(httpPost, cb);
}

public static void main(String[] args) throws Exception {
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
    try {
        httpclient.start();
        HttpHost proxy = new HttpHost("someproxy", 8080);
        RequestConfig config = RequestConfig.custom().setProxy(proxy).build();
        HttpGet request = new HttpGet("https://issues.apache.org/");
        request.setConfig(config);
        Future<HttpResponse> future = httpclient.execute(request, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

public static void main(final String[] args) throws Exception {
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
    try {
        httpclient.start();
        Future<Boolean> future = httpclient.execute(HttpAsyncMethods.createGet("http://localhost:8080/"),
                new MyResponseConsumer(), null);
        Boolean result = future.get();
        if (result != null && result.booleanValue()) {
            System.out.println("Request successfully executed");
        } else {
            System.out.println("Request failed");
        }
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
    System.out.println("Done");
}

public static void main(final String[] args) throws Exception {
    CloseableHttpPipeliningClient httpclient = HttpAsyncClients.createPipelining();
    try {
        httpclient.start();

        HttpHost targetHost = new HttpHost("localhost", 8080);
        HttpGet[] resquests = { new HttpGet("/docs/index.html"), new HttpGet("/docs/introduction.html"),
                new HttpGet("/docs/setup.html"), new HttpGet("/docs/config/index.html") };

        Future<List<HttpResponse>> future = httpclient.execute(targetHost, Arrays.<HttpRequest>asList(resquests),
                null);
        List<HttpResponse> responses = future.get();
        System.out.println(responses);

        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
    System.out.println("Done");
}

public static void main(String[] args) throws Exception {
    CredentialsProvider credsProvider = new BasicCredentialsProvider();
    credsProvider.setCredentials(new AuthScope("localhost", 443),
            new UsernamePasswordCredentials("username", "password"));
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setDefaultCredentialsProvider(credsProvider)
            .build();
    try {
        HttpGet httpget = new HttpGet("http://localhost/");

        System.out.println("Executing request " + httpget.getRequestLine());
        Future<HttpResponse> future = httpclient.execute(httpget, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

public static void main(final String[] args) throws Exception {
    CloseableHttpPipeliningClient httpclient = HttpAsyncClients.createPipelining();
    try {
        httpclient.start();

        HttpHost targetHost = new HttpHost("localhost", 8080);
        HttpGet[] resquests = { new HttpGet("/docs/index.html"), new HttpGet("/docs/introduction.html"),
                new HttpGet("/docs/setup.html"), new HttpGet("/docs/config/index.html") };

        List<MyRequestProducer> requestProducers = new ArrayList<MyRequestProducer>();
        List<MyResponseConsumer> responseConsumers = new ArrayList<MyResponseConsumer>();
        for (HttpGet request : resquests) {
            requestProducers.add(new MyRequestProducer(targetHost, request));
            responseConsumers.add(new MyResponseConsumer(request));
        }

        Future<List<Boolean>> future = httpclient.execute(targetHost, requestProducers, responseConsumers, null);
        future.get();
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
    System.out.println("Done");
}

public static void main(String[] args) throws Exception {
    CredentialsProvider credsProvider = new BasicCredentialsProvider();
    credsProvider.setCredentials(new AuthScope("someproxy", 8080),
            new UsernamePasswordCredentials("username", "password"));
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setDefaultCredentialsProvider(credsProvider)
            .build();
    try {
        httpclient.start();
        HttpHost proxy = new HttpHost("someproxy", 8080);
        RequestConfig config = RequestConfig.custom().setProxy(proxy).build();
        HttpGet httpget = new HttpGet("https://issues.apache.org/");
        httpget.setConfig(config);
        Future<HttpResponse> future = httpclient.execute(httpget, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

private void open() {
   if (open) {
      // Ignore
      return;
   }

   final HttpAsyncClientBuilder builder = HttpAsyncClients.custom()
         .setMaxConnPerRoute(1000)
         .setMaxConnTotal(1000)
         .setKeepAliveStrategy(new DefaultConnectionKeepAliveStrategy())
         .setThreadFactory(threadFactory);
   if (credentialsProvider != null) {
      builder.setDefaultCredentialsProvider(credentialsProvider);
   }

   httpclient = builder.build();
   httpclient.start();
   this.open = true;
}

public final static HttpResponse get(String url, List<NameValuePair> parameters) throws Throwable {
    CloseableHttpAsyncClient client = HttpAsyncClients.createDefault();

    StringBuffer sb = new StringBuffer("?");
    for (NameValuePair pair : parameters) {
        sb.append(pair.getName());
        sb.append("=");
        sb.append(URLEncoder.encode(pair.getValue(), "UTF-8"));
        sb.append("&");
    }

    client.start();
    final HttpGet httpGet = new HttpGet(url + sb.toString());
    httpGet.removeHeaders("X-FORWARDED-FOR");
    httpGet.setHeader("X-FORWARDED-FOR", Environment.LOCAL_IP_ADDR);

    logger.debug("-> GET " + (url + sb.toString()));

    Future<HttpResponse> future = client.execute(httpGet, null);
    HttpResponse resp = future.get();

    return resp;
}

public final static HttpResponse post(String url, List<NameValuePair> parameters) throws Throwable {
    CloseableHttpAsyncClient client = HttpAsyncClients.createDefault();

    client.start();
    final HttpPost httpPost = new HttpPost(url);
    httpPost.removeHeaders("X-FORWARDED-FOR");
    httpPost.setHeader("X-FORWARDED-FOR", Environment.LOCAL_IP_ADDR);
    UrlEncodedFormEntity encodedFormEntity = new UrlEncodedFormEntity(parameters, "UTF-8");
    httpPost.setEntity(encodedFormEntity);

    logger.debug("-> POST " + url + " Parameters " + JsonUtil.toString(parameters));

    Future<HttpResponse> future = client.execute(httpPost, null);
    HttpResponse resp = future.get();

    return resp;
}

public final static HttpResponse post(String url, Object model) throws Throwable {
    CloseableHttpAsyncClient client = HttpAsyncClients.createDefault();

    client.start();
    final HttpPost httpPost = new HttpPost(url);
    httpPost.removeHeaders("X-FORWARDED-FOR");
    httpPost.setHeader("X-FORWARDED-FOR", Environment.LOCAL_IP_ADDR);
    StringEntity stringEntity = new StringEntity(JsonUtil.toString(model), ENCODING);
    stringEntity.setContentType(CONTENT_TYPE);
    httpPost.setEntity(stringEntity);

    logger.debug("-> POST " + url + " Parameters " + JsonUtil.toString(model));

    Future<HttpResponse> future = client.execute(httpPost, null);
    HttpResponse resp = future.get();

    return resp;
}

public final static HttpResponse post(String url, Object model, HttpEntity httpEntity) throws Throwable {
    CloseableHttpAsyncClient client = HttpAsyncClients.createDefault();

    client.start();
    final HttpPost httpPost = new HttpPost(url);
    httpPost.removeHeaders("X-FORWARDED-FOR");
    httpPost.setHeader("X-FORWARDED-FOR", Environment.LOCAL_IP_ADDR);
    httpPost.setEntity(httpEntity);

    logger.debug("-> POST " + url + " httpEntity  Parameters " + JsonUtil.toString(model));

    Future<HttpResponse> future = client.execute(httpPost, null);
    HttpResponse resp = future.get();

    return resp;
}

/**
 * Initializes and returns the httpClient with NoopHostnameVerifier
 * 
 * @return CloseableHttpAsyncClient
 */
@Override
public CloseableHttpAsyncClient getHttpClient() {
    // Trust own CA and all self-signed certs
    SSLContext sslcontext = NonValidatingSSLSocketFactory.getSSLContext();
    // Allow TLSv1 protocol only

    SSLIOSessionStrategy sslSessionStrategy = new SSLIOSessionStrategy(sslcontext, new String[] { "TLSv1" }, null,
            new NoopHostnameVerifier());
    List<Header> headers = LogInsightClient.getDefaultHeaders();

    asyncHttpClient = HttpAsyncClients.custom().setSSLStrategy(sslSessionStrategy).setDefaultHeaders(headers)
            .build();
    asyncHttpClient.start();

    return asyncHttpClient;
}

/**
 * Creates an asynchronous HTTP client configuration with default timeouts.
 *
 * @see #newHttpAsyncClient(boolean)
 */
protected static CloseableHttpAsyncClient newHttpAsyncClient(boolean useSSL) {
    RequestConfig requestConfig = RequestConfig.custom().setSocketTimeout(DEFAULT_SOCKET_TIMEOUT)
            .setConnectTimeout(DEFAULT_CONNECT_TIMEOUT).build();

    HttpAsyncClientBuilder builder = HttpAsyncClients.custom();

    if (useSSL) {
        try {
            SSLContext context = SSLContext.getInstance("SSL");
            context.init(null, new TrustManager[]{new TrustAllX509Manager()}, new SecureRandom());

            SSLIOSessionStrategy strategy = new SSLIOSessionStrategy(context,
                    SSLIOSessionStrategy.getDefaultHostnameVerifier());

            builder.setSSLStrategy(strategy);
        } catch (Exception e) {
            log.error("Failed initializing SSL context! Skipped.", e);
        }
    }

    return builder.setDefaultRequestConfig(requestConfig).build();
}

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

@Override
public CloseableHttpAsyncClient generateClient ()
{
   CredentialsProvider credsProvider = new BasicCredentialsProvider();
   credsProvider.setCredentials(new AuthScope (AuthScope.ANY),
           new UsernamePasswordCredentials(serviceUser, servicePass));
   RequestConfig rqconf = RequestConfig.custom()
         .setCookieSpec(CookieSpecs.DEFAULT)
         .setSocketTimeout(Timeouts.SOCKET_TIMEOUT)
         .setConnectTimeout(Timeouts.CONNECTION_TIMEOUT)
         .setConnectionRequestTimeout(Timeouts.CONNECTION_REQUEST_TIMEOUT)
         .build();
   CloseableHttpAsyncClient res = HttpAsyncClients.custom ()
         .setDefaultCredentialsProvider (credsProvider)
         .setDefaultRequestConfig(rqconf)
         .build ();
   res.start ();
   return res;
}

@Override
public CloseableHttpAsyncClient generateClient ()
{
   CredentialsProvider credsProvider = new BasicCredentialsProvider();
   credsProvider.setCredentials(new AuthScope (AuthScope.ANY),
            new UsernamePasswordCredentials(username, password));
   RequestConfig rqconf = RequestConfig.custom()
         .setCookieSpec(CookieSpecs.DEFAULT)
         .setSocketTimeout(Timeouts.SOCKET_TIMEOUT)
         .setConnectTimeout(Timeouts.CONNECTION_TIMEOUT)
         .setConnectionRequestTimeout(Timeouts.CONNECTION_REQUEST_TIMEOUT)
         .build();
   CloseableHttpAsyncClient res = HttpAsyncClients.custom ()
         .setDefaultCredentialsProvider (credsProvider)
         .setDefaultRequestConfig(rqconf)
         .build ();
   res.start ();
   return res;
}

private CloseableHttpAsyncClient getAsyncProxyHttpClient(URI proxyUri) {
  LOG.info("Creating async proxy http client");
  PoolingNHttpClientConnectionManager cm = getAsyncConnectionManager();
  HttpHost proxy = new HttpHost(proxyUri.getHost(), proxyUri.getPort());

  HttpAsyncClientBuilder clientBuilder = HttpAsyncClients.custom();
  if (cm != null) {
    clientBuilder = clientBuilder.setConnectionManager(cm);
  }

  if (proxy != null) {
    clientBuilder = clientBuilder.setProxy(proxy);
  }
  clientBuilder = setRedirects(clientBuilder);
  return clientBuilder.build();
}

public static void main(String[] args)throws Exception {
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
    try {
        httpclient.start();
        HttpHost proxy = new HttpHost("someproxy", 8080);
        RequestConfig config = RequestConfig.custom()
                .setProxy(proxy)
                .build();
        HttpGet request = new HttpGet("https://issues.apache.org/");
        request.setConfig(config);
        Future<HttpResponse> future = httpclient.execute(request, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

public static void main(final String[] args) throws Exception {
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
    try {
        httpclient.start();
        Future<Boolean> future = httpclient.execute(
                HttpAsyncMethods.createGet("http://localhost:8080/"),
                new MyResponseConsumer(), null);
        Boolean result = future.get();
        if (result != null && result.booleanValue()) {
            System.out.println("Request successfully executed");
        } else {
            System.out.println("Request failed");
        }
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
    System.out.println("Done");
}

public static void main(final String[] args) throws Exception {
    CloseableHttpPipeliningClient httpclient = HttpAsyncClients.createPipelining();
    try {
        httpclient.start();

        HttpHost targetHost = new HttpHost("localhost", 8080);
        HttpGet[] resquests = {
                new HttpGet("/docs/index.html"),
                new HttpGet("/docs/introduction.html"),
                new HttpGet("/docs/setup.html"),
                new HttpGet("/docs/config/index.html")
        };

        Future<List<HttpResponse>> future = httpclient.execute(targetHost,
                Arrays.<HttpRequest>asList(resquests), null);
        List<HttpResponse> responses = future.get();
        System.out.println(responses);

        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
    System.out.println("Done");
}

public static void main(String[] args) throws Exception {
    CredentialsProvider credsProvider = new BasicCredentialsProvider();
    credsProvider.setCredentials(
            new AuthScope("localhost", 443),
            new UsernamePasswordCredentials("username", "password"));
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom()
            .setDefaultCredentialsProvider(credsProvider)
            .build();
    try {
        HttpGet httpget = new HttpGet("http://localhost/");

        System.out.println("Executing request " + httpget.getRequestLine());
        Future<HttpResponse> future = httpclient.execute(httpget, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

public static void main(String[] args)throws Exception {
    CredentialsProvider credsProvider = new BasicCredentialsProvider();
    credsProvider.setCredentials(
            new AuthScope("someproxy", 8080),
            new UsernamePasswordCredentials("username", "password"));
    CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom()
            .setDefaultCredentialsProvider(credsProvider)
            .build();
    try {
        httpclient.start();
        HttpHost proxy = new HttpHost("someproxy", 8080);
        RequestConfig config = RequestConfig.custom()
                .setProxy(proxy)
                .build();
        HttpGet httpget = new HttpGet("https://issues.apache.org/");
        httpget.setConfig(config);
        Future<HttpResponse> future = httpclient.execute(httpget, null);
        HttpResponse response = future.get();
        System.out.println("Response: " + response.getStatusLine());
        System.out.println("Shutting down");
    } finally {
        httpclient.close();
    }
}

private QueryExecutor(String userAgent, ObjectMapper mapper, HttpHost proxy)
{
    checkNotNull(userAgent, "userAgent is null");
    checkNotNull(mapper, "mapper is null");

    this.userAgent = userAgent;
    this.mapper = mapper;

    HttpClientBuilder builder = HttpClients.custom();
    HttpAsyncClientBuilder asyncBuilder = HttpAsyncClients.custom();

    if (proxy != null) {
        DefaultProxyRoutePlanner routePlanner = new DefaultProxyRoutePlanner(proxy);
        builder.setRoutePlanner(routePlanner);
        asyncBuilder.setRoutePlanner(routePlanner);
    }

    this.httpClient = asyncBuilder.build();
    this.httpClient.start();
}

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

/**
 * Creates asynchronous Apache HTTP client.
 *
 * @param settings
 *         settings to use to create client.
 * @param conf
 *         configuration related to async connection.
 * @return Instance of {@link CloseableHttpAsyncClient}.
 */
private CloseableHttpAsyncClient createClient(HttpSettings settings, ApacheHttpClientConfiguration conf) {
    IOReactorConfig ioReactor = IOReactorConfig.custom().setIoThreadCount(conf.getMaxThreadCount()).build();
    HttpAsyncClientBuilder httpClientBuilder = HttpAsyncClients.custom()
            .useSystemProperties()
                    // allow POST redirects
            .setRedirectStrategy(new LaxRedirectStrategy()).setMaxConnTotal(conf.getMaxTotalConnectionCount()).setMaxConnPerRoute(conf.getMaxRouteConnectionCount()).setDefaultIOReactorConfig(ioReactor)
            .setKeepAliveStrategy(new DefaultConnectionKeepAliveStrategy()).setDefaultRequestConfig(createDefaultRequestConfig(settings));
    if (settings.getProxyUrl() != null) {
        DefaultProxyRoutePlanner routePlanner = createProxyRoutePlanner(settings, httpClientBuilder);
        httpClientBuilder.setRoutePlanner(routePlanner);
    }
    CloseableHttpAsyncClient httpClient = httpClientBuilder.build();
    httpClient.start();
    return httpClient;
}

@Override
public void transactionMarker() throws Exception {
    CloseableHttpAsyncClient httpClient = HttpAsyncClients.createDefault();
    httpClient.start();
    HttpHost httpHost = new HttpHost("localhost", getPort());
    HttpGet httpGet = new HttpGet("/hello2");
    SimpleFutureCallback callback = new SimpleFutureCallback();
    Future<HttpResponse> future = httpClient.execute(httpHost, httpGet, callback);
    callback.latch.await();
    httpClient.close();
    int responseStatusCode = future.get().getStatusLine().getStatusCode();
    if (responseStatusCode != 200) {
        throw new IllegalStateException(
                "Unexpected response status code: " + responseStatusCode);
    }
}

@Override
public void transactionMarker() throws Exception {
    CloseableHttpAsyncClient httpClient = HttpAsyncClients.createDefault();
    httpClient.start();
    HttpHost httpHost = new HttpHost("localhost", getPort());
    HttpPost httpPost = new HttpPost("/hello4");
    SimpleFutureCallback callback = new SimpleFutureCallback();
    Future<HttpResponse> future = httpClient.execute(httpHost, httpPost, callback);
    callback.latch.await();
    httpClient.close();
    int responseStatusCode = future.get().getStatusLine().getStatusCode();
    if (responseStatusCode != 200) {
        throw new IllegalStateException(
                "Unexpected response status code: " + responseStatusCode);
    }
}

@Override
public void init(ZibaseDeviceConfiguration configuration) {
    this.configuration = configuration;
    try {
        // init HTTP client
        SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(null, (chain, authType) -> true).build();
        httpClient = HttpAsyncClients.custom().setSSLContext(sslContext).build();
        httpClient.start();

        // configure Json parser
        mapper = new ObjectMapper();
        mapper.getFactory().configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);

        token = fetchToken();
        // TODO: fetch Zibase devices after init
    } catch (NoSuchAlgorithmException | KeyManagementException | KeyStoreException e) {
        logger.error("Can't initialize SSL engine", e);
    }
}

private static synchronized CacheLoader<String, String> getCacheLoader() {
    return new CacheLoader<String, String>() {
        @Override
        public String load(String key) throws Exception {
            log.trace("URI=[{}] 의 웹 컨텐츠를 비동기 방식으로 다운로드 받아 캐시합니다.", key);

            String responseStr = "";
            CloseableHttpAsyncClient httpClient = HttpAsyncClients.createDefault(); //new DefaultHttpAsyncClient();
            try {
                httpClient.start();
                HttpGet request = new HttpGet(key);
                Future<HttpResponse> future = httpClient.execute(request, null);

                HttpResponse response = future.get();
                responseStr = EntityUtils.toString(response.getEntity(), Charsets.UTF_8.toString());

                if (log.isDebugEnabled())
                    log.debug("URI=[{}]로부터 웹 컨텐츠를 다운로드 받았습니다. responseStr=[{}]",
                              key, StringTool.ellipsisChar(responseStr, 80));
            } finally {
                httpClient.close();
            }
            return responseStr;
        }
    };
}

@Override
public void init(QQContext context) throws QQException {
    super.init(context);
    RequestConfig requestConfig = RequestConfig.custom()
            .setSocketTimeout(QQConstants.HTTP_TIME_OUT)
            .setConnectTimeout(QQConstants.HTTP_TIME_OUT)
            .build();
    SSLContext sslContext = new QQSSLSocketFactory().getSSLContext();
    SSLContext.setDefault(sslContext);
    asyncHttpClient = HttpAsyncClients.custom()
            .setDefaultRequestConfig(requestConfig)
            .setRedirectStrategy(new QQDefaultRedirectStrategy())
            .build();
    asyncHttpClient.start();
    cookieJar = new QQHttpCookieJar();
}

@Before
public void before() throws Exception {
  tsdb = PowerMockito.mock(TSDB.class);
  config = new Config(false);
  connection_manager = mock(PoolingNHttpClientConnectionManager.class);
  client_builder = mock(HttpAsyncClientBuilder.class);
  client = mock(CloseableHttpAsyncClient.class);
  ts_meta_schema = mock(TSMetaSchema.class);
  uid_meta_schema = mock(UIDMetaSchema.class);
  annotation_schema = mock(AnnotationSchema.class);

  config.overrideConfig("tsd.search.elasticsearch.host", "localhost:9200");

  when(tsdb.getConfig()).thenReturn(config);

  PowerMockito.mockStatic(HttpAsyncClients.class);
  when(HttpAsyncClients.custom()).thenReturn(client_builder);

  PowerMockito.whenNew(PoolingNHttpClientConnectionManager.class)
    .withAnyArguments().thenReturn(connection_manager);
  when(client_builder.build()).thenReturn(client);
}

private CloseableHttpAsyncClient createHttpClient() {

        final RequestConfig.Builder requestConfigBuilder = RequestConfig.custom();
        requestConfigBuilder.setConnectionRequestTimeout(TIMEOUT_CONNECTION_REQUEST);
        requestConfigBuilder.setConnectTimeout(TIMEOUT_CONNECT);
        requestConfigBuilder.setSocketTimeout(TIMEOUT_SOCKET);

        if (this.mProxyHost != null) {
            requestConfigBuilder.setProxy(this.mProxyHost);
        }

        final HttpAsyncClientBuilder clientBuilder = HttpAsyncClients.custom();

        final RequestConfig requestConfig = requestConfigBuilder.build();
        clientBuilder.setDefaultRequestConfig(requestConfig);

        if (this.mProxyHost != null && !StringUtils.isEmpty(this.mProxyUsername)) {
            final AuthScope authScope = new AuthScope(this.mProxyHost);
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            final Credentials credentials = new UsernamePasswordCredentials(this.mProxyUsername, this.mProxyPassword);
            credentialsProvider.setCredentials(authScope, credentials);
            clientBuilder.setDefaultCredentialsProvider(credentialsProvider);
        }

        return clientBuilder.build();
    }

public static CloseableHttpAsyncClient create(RequestConfig requestConfig) {
    HttpAsyncClientBuilder builder = HttpAsyncClients.custom();
    builder.setConnectionManager(connManager)
            .setDefaultCookieStore(cookieStore)
            .setDefaultCredentialsProvider(credentialsProvider);

    if (null != requestConfig) {
        return builder.setDefaultRequestConfig(requestConfig).build();
    } else {
        return builder.setDefaultRequestConfig(defaultRequestConfig)
                .build();
    }
}

public HttpPostDeliverService(final String postUrl, final int connectTimeout, final int soTimeout) {
  httpClient = HttpAsyncClients.createDefault();
  httpClient.start();

  httpPost = new HttpPost(postUrl);
  final RequestConfig requestConfig =
      RequestConfig.custom().setConnectTimeout(connectTimeout).setSocketTimeout(soTimeout).build();
  httpPost.setConfig(requestConfig);

  httpPost.setHeader("Content-type", "application/json");
  httpPost.setHeader("Content-Type", "text/html;charset=UTF-8");
}

/** An InterruptibleHttpClient using {@code HttpAsyncClients.createDefault()}
 * as HttpAsyncClientProducer. */
public InterruptibleHttpClient ()
{
   clientProducer = new HttpAsyncClientProducer ()
   {
      @Override
      public CloseableHttpAsyncClient generateClient ()
      {
         CloseableHttpAsyncClient res = HttpAsyncClients.createDefault ();
         res.start ();
         return res;
      }
   };
}