org.apache.http.client.methods.HttpExecutionAware#org.apache.http.conn.routing.HttpRoute源码实例Demo

/**
 * {@inheritDoc}
 */
@Override
public HttpRoute determineRoute(final HttpHost host, final HttpRequest request,
        final HttpContext context) throws HttpException {

    Boolean ignoreNoProxy = (Boolean) context.getAttribute(IGNORE_NO_PROXY);
    URI requestUri = (URI) context.getAttribute(REQUEST_URI);

    if (proxyHost != null &&
            (Boolean.TRUE.equals(ignoreNoProxy) || useProxyFor(requestUri))) {
        if (log.isDebugEnabled()) {
            log.debug("Using proxy: " + proxyHost);
        }
        return super.determineRoute(host, request, context);
    }
    if (log.isDebugEnabled()) {
        log.debug("Using a direct connection (no proxy)");
    }
    // Return direct route
    return new HttpRoute(host);
}
 

@Advice.OnMethodEnter(suppress = Throwable.class)
private static void onBeforeExecute(@Advice.Argument(0) HttpRoute route,
                                    @Advice.Argument(1) HttpRequestWrapper request,
                                    @Advice.Local("span") Span span) {
    if (tracer == null || tracer.getActive() == null) {
        return;
    }
    final AbstractSpan<?> parent = tracer.getActive();
    span = HttpClientHelper.startHttpClientSpan(parent, request.getMethod(), request.getURI(), route.getTargetHost().getHostName());
    TextHeaderSetter<HttpRequest> headerSetter = headerSetterHelperClassManager.getForClassLoaderOfClass(HttpRequest.class);
    TextHeaderGetter<HttpRequest> headerGetter = headerGetterHelperClassManager.getForClassLoaderOfClass(HttpRequest.class);
    if (span != null) {
        span.activate();
        if (headerSetter != null) {
            span.propagateTraceContext(request, headerSetter);
        }
    } else if (headerGetter != null && !TraceContext.containsTraceContextTextHeaders(request, headerGetter)
        && headerSetter != null && parent != null) {
        // re-adds the header on redirects
        parent.propagateTraceContext(request, headerSetter);
    }
}
 

static Tags generateTagsForRoute(HttpContext context) {
    String targetScheme = "UNKNOWN";
    String targetHost = "UNKNOWN";
    String targetPort = "UNKNOWN";
    Object routeAttribute = context.getAttribute("http.route");
    if (routeAttribute instanceof HttpRoute) {
        HttpHost host = ((HttpRoute) routeAttribute).getTargetHost();
        targetScheme = host.getSchemeName();
        targetHost = host.getHostName();
        targetPort = String.valueOf(host.getPort());
    }
    return Tags.of(
            "target.scheme", targetScheme,
            "target.host", targetHost,
            "target.port", targetPort
    );
}
 

@Override
public HttpRoute determineRoute(HttpHost host, HttpRequest request, HttpContext context) throws HttpException {
    try {
        ClientCredential clientCredential = binding.getClientCredentialProvider().getClientCredential(binding.map(request));
        String hostname = clientCredential.getHost();
        int port = -1;
        final int pos = hostname.lastIndexOf(":");
        if (pos > 0) {
            try {
                port = Integer.parseInt(hostname.substring(pos + 1));
            } catch (NumberFormatException ex) {
                throw new IllegalArgumentException("Host contains invalid port number: " + hostname);
            }
            hostname = hostname.substring(0, pos);
        }
        HttpHost target = new HttpHost(hostname, port, "https");
        return super.determineRoute(target, request, context);
    } catch (NoMatchingCredentialException e) {
        throw new RuntimeException(e);
    }
}
 

@Override
public ManagedHttpClientConnection create(final HttpRoute route, final ConnectionConfig config) {
    final ConnectionConfig cconfig = config != null ? config : ConnectionConfig.DEFAULT;
    CharsetDecoder chardecoder = null;
    CharsetEncoder charencoder = null;
    final Charset charset = cconfig.getCharset();
    final CodingErrorAction malformedInputAction = cconfig.getMalformedInputAction() != null ? cconfig.getMalformedInputAction() : CodingErrorAction.REPORT;
    final CodingErrorAction unmappableInputAction = cconfig.getUnmappableInputAction() != null ? cconfig.getUnmappableInputAction()
            : CodingErrorAction.REPORT;
    if (charset != null) {
        chardecoder = charset.newDecoder();
        chardecoder.onMalformedInput(malformedInputAction);
        chardecoder.onUnmappableCharacter(unmappableInputAction);
        charencoder = charset.newEncoder();
        charencoder.onMalformedInput(malformedInputAction);
        charencoder.onUnmappableCharacter(unmappableInputAction);
    }
    final String id = "http-outgoing-" + Long.toString(COUNTER.getAndIncrement());
    return new TracingManagedHttpClientConnection(id, cconfig.getBufferSize(), cconfig.getFragmentSizeHint(), chardecoder, charencoder,
            cconfig.getMessageConstraints(), incomingContentStrategy, outgoingContentStrategy, requestWriterFactory, responseParserFactory, logFunc);
}
 

@Test
public void routePlannerIsInvoked() throws Exception {
    mockProxyServer.resetToDefaultMappings();
    mockProxyServer.addStubMapping(WireMock.any(urlPathEqualTo("/"))
                                           .willReturn(aResponse().proxiedFrom("http://localhost:" + mockServer.port()))
                                           .build());

    SdkHttpClient client = ApacheHttpClient.builder()
                                           .httpRoutePlanner(
                                               (host, request, context) ->
                                                   new HttpRoute(
                                                       new HttpHost("localhost", mockProxyServer.httpsPort(), "https")
                                                   )
                                           )
                                           .buildWithDefaults(AttributeMap.builder()
                                                                          .put(TRUST_ALL_CERTIFICATES, Boolean.TRUE)
                                                                          .build());

    testForResponseCodeUsingHttps(client, HttpURLConnection.HTTP_OK);

    mockProxyServer.verify(1, RequestPatternBuilder.allRequests());
}
 

@Override public CloseableHttpResponse execute(HttpRoute route,
  org.apache.http.client.methods.HttpRequestWrapper req,
  HttpClientContext context, HttpExecutionAware execAware)
  throws IOException, HttpException {
  HttpRequestWrapper request = new HttpRequestWrapper(req, context.getTargetHost());
  Span span = tracer.nextSpan(httpSampler, request);
  context.setAttribute(Span.class.getName(), span);

  CloseableHttpResponse response = null;
  Throwable error = null;
  try (SpanInScope ws = tracer.withSpanInScope(span)) {
    return response = protocolExec.execute(route, req, context, execAware);
  } catch (Throwable e) {
    error = e;
    throw e;
  } finally {
    handler.handleReceive(new HttpResponseWrapper(response, context, error), span);
  }
}
 

@Override
public CloseableHttpResponse execute(HttpRoute route, HttpRequestWrapper request, HttpClientContext clientContext,
        HttpExecutionAware execAware) throws IOException, HttpException {
    Proxy proxy = (Proxy) clientContext.getAttribute(VSCrawlerConstant.VSCRAWLER_AVPROXY_KEY);
    if (proxy != null) {
        proxy.recordUsage();
    }
    try {
        return delegate.execute(route, request, clientContext, execAware);
    } catch (IOException ioe) {
        if (proxy != null) {
            proxy.recordFailed();
        }
        throw ioe;
    }
}
 

@Override
protected NHttpClientConnectionManager getAsyncConnectionManager() {

    PoolingNHttpClientConnectionManager connectionManager = createUnconfiguredPoolingNHttpClientConnectionManager();

    HttpClientConfig httpClientConfig = this.wrappedHttpClientConfig.getHttpClientConfig();

    final Integer maxTotal = httpClientConfig.getMaxTotalConnection();
    if (maxTotal != null) {
        connectionManager.setMaxTotal(maxTotal);
    }
    final Integer defaultMaxPerRoute = httpClientConfig.getDefaultMaxTotalConnectionPerRoute();
    if (defaultMaxPerRoute != null) {
        connectionManager.setDefaultMaxPerRoute(defaultMaxPerRoute);
    }
    final Map<HttpRoute, Integer> maxPerRoute = httpClientConfig.getMaxTotalConnectionPerRoute();
    for (Map.Entry<HttpRoute, Integer> entry : maxPerRoute.entrySet()) {
        connectionManager.setMaxPerRoute(entry.getKey(), entry.getValue());
    }

    return connectionManager;
}
 

@Test
public void getAsyncConnectionManagerConfiguresMaxTotalPerRouteIfConfigured() {

    // given
    HttpClientConfig.Builder config = createDefaultTestHttpClientConfigBuilder();

    HttpRoute expectedHttpRoute = new HttpRoute(new HttpHost("localhost"));
    int expectedMaxTotalConnection = random.nextInt(100) + 10;
    config.maxTotalConnectionPerRoute(expectedHttpRoute, expectedMaxTotalConnection);

    WrappedHttpClientConfig.Builder builder = createDefaultTestWrappedHttpClientConfigBuilder(config.build());
    ExtendedJestClientFactory factory = spy(new ExtendedJestClientFactory(builder.build()));

    PoolingNHttpClientConnectionManager mockedNHttpConnectionManager = mock(PoolingNHttpClientConnectionManager.class);
    when(factory.createUnconfiguredPoolingNHttpClientConnectionManager())
            .thenReturn(mockedNHttpConnectionManager);

    // when
    factory.getAsyncConnectionManager();

    // then
    verify(mockedNHttpConnectionManager).setMaxPerRoute(eq(expectedHttpRoute), eq(expectedMaxTotalConnection));

}
 

public static HttpClient generateHttpClient(final int maxTotalConnections,
    final int maxTotalConnectionsPerRoute, int connTimeout) {
  HttpParams params = new BasicHttpParams();
  ConnManagerParams.setMaxTotalConnections(params, maxTotalConnections);
  ConnManagerParams.setMaxConnectionsPerRoute(params, new ConnPerRoute() {
    @Override
    public int getMaxForRoute(HttpRoute route) {
      return maxTotalConnectionsPerRoute;
    }
  });
  HttpConnectionParams
      .setConnectionTimeout(params, connTimeout);
  SchemeRegistry schemeRegistry = new SchemeRegistry();
  schemeRegistry.register(
      new Scheme("http", PlainSocketFactory.getSocketFactory(), 80));
  SSLSocketFactory sslSocketFactory = SSLSocketFactory.getSocketFactory();
  sslSocketFactory.setHostnameVerifier(SSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
  schemeRegistry.register(new Scheme("https", sslSocketFactory, 443));
  ClientConnectionManager conMgr = new ThreadSafeClientConnManager(params, schemeRegistry);
  return new DefaultHttpClient(conMgr, params);
}
 

private static HttpClientConnectionManager createDefaultConnectionManager(
    HttpClientConnectionOperator operator,
    HttpConnectionFactory<HttpRoute, ManagedHttpClientConnection> connectionFactory
) {
  PoolingHttpClientConnectionManager manager = new PoolingHttpClientConnectionManager(
      operator,
      connectionFactory,
      -1,
      TimeUnit.MILLISECONDS
  );

  manager.setMaxTotal(3000);
  manager.setDefaultMaxPerRoute(1500);

  return manager;
}
 

@Override
public ConnectionRequest requestConnection(HttpRoute route, Object state) {
  return new ConnectionRequest() {

    @Override
    public boolean cancel() {
      // Nothing to do.
      return false;
    }

    @Override
    public HttpClientConnection get(final long timeout, final TimeUnit timeUnit) {
      return connectionFactory.create(route, connectionConfig);
    }
  };
}
 

@Override
public void connect(
    HttpClientConnection connection,
    HttpRoute route,
    int connectTimeout,
    HttpContext context
) throws IOException {
  HttpHost host;

  if (route.getProxyHost() != null) {
    host = route.getProxyHost();
  } else {
    host = route.getTargetHost();
  }

  InetSocketAddress localAddress = route.getLocalSocketAddress();

  ManagedHttpClientConnection managed = (ManagedHttpClientConnection) connection;
  this.connectionOperator.connect(managed, host, localAddress, connectTimeout, this.socketConfig, context);
}
 

private CloseableHttpClient createHttpClient(String hostname, int port) {
    ConnectionSocketFactory plainsf = PlainConnectionSocketFactory.getSocketFactory();
    LayeredConnectionSocketFactory sslsf = SSLConnectionSocketFactory.getSocketFactory();
    Registry<ConnectionSocketFactory> registry = RegistryBuilder.<ConnectionSocketFactory> create()
            .register("http", plainsf).register("https", sslsf).build();
    PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager(registry);
    // 将最大连接数增加
    cm.setMaxTotal(maxTotal);
    // 将每个路由基础的连接增加
    cm.setDefaultMaxPerRoute(maxPerRoute);
    HttpHost httpHost = new HttpHost(hostname, port);
    // 将目标主机的最大连接数增加
    cm.setMaxPerRoute(new HttpRoute(httpHost), maxRoute);
    // 请求重试处理
    return HttpClients.custom().setConnectionManager(cm).setRetryHandler(httpRequestRetryHandler).build();
}
 

/**
 * @param timeEntity                how often the {@link #monitor()} check needs to be executed
 * @param httpRequest               http request that will be called at regular intervals
 * @param pingTimeoutInMilliseconds timeout in milliseconds for http request execution (ping response)
 * @param pingWindowSize            rolling window frame, which needs to be maintained
 * @param maxFailures               maximum failures allowed in the rolling window frame
 * @param host                      host name (could be localhost)
 * @param port                      port
 */
public PingCheckMonitor(TimeEntity timeEntity,
                        HttpRequest httpRequest,
                        Integer pingTimeoutInMilliseconds,
                        Integer pingWindowSize,
                        Integer maxFailures,
                        String host,
                        Integer port) {
    super(PingCheckMonitor.class.getSimpleName(), timeEntity);
    this.httpRequest = httpRequest;
    this.pingTimeoutInMilliseconds = pingTimeoutInMilliseconds;
    this.host = host;
    this.port = port;
    this.rollingWindowHealthQueue = new RollingWindowHealthQueue(pingWindowSize, maxFailures);
    this.executorService = Executors.newSingleThreadExecutor();
    PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
    connectionManager.setMaxPerRoute(new HttpRoute(new HttpHost(host, port)), 2);
    this.httpClient = HttpClients.custom()
            .setConnectionManager(connectionManager)
            .build();
}
 

protected void rewriteRequestURI(
        final RequestWrapper request,
        final HttpRoute route) throws ProtocolException {
    try {

        URI uri = request.getURI();
        if (route.getProxyHost() != null && !route.isTunnelled()) {
            // Make sure the request URI is absolute
            if (!uri.isAbsolute()) {
                HttpHost target = route.getTargetHost();
                uri = URIUtils.rewriteURI(uri, target);
                request.setURI(uri);
            }
        } else {
            // Make sure the request URI is relative
            if (uri.isAbsolute()) {
                uri = URIUtils.rewriteURI(uri, null);
                request.setURI(uri);
            }
        }

    } catch (URISyntaxException ex) {
        throw new ProtocolException("Invalid URI: "
        		+ request.getRequestLine().getUri(), ex);
    }
}
 

private void processOnRetry(RetryEvent event, Duration waitInterval,
                            HttpRoutePlanner planer, OnRetryConsumer onRetryConsumer) {

    InetAddress targetAddress = null;
    if (planer instanceof RoundRobinRouterPlaner) {
        // inform planer about error on last roue
        HttpRoute httpRoute = ((RoundRobinRouterPlaner)planer).lastRouteCauseError();
        targetAddress = Try.of(() -> httpRoute.getTargetHost().getAddress()).getOrElse((InetAddress) null);
    } else if (proxy != null) {
        targetAddress = Try.of(() -> InetAddress.getByName(proxy.getHost())).getOrElse((InetAddress) null);
    }

    // inform caller about retry
    if (onRetryConsumer != null) {
        onRetryConsumer.onRetry(targetAddress, event.getNumberOfRetryAttempts(),
                waitInterval, event.getLastThrowable());
    }
}
 

@Test
public void shouldReturnTheSameAddressForSequentialCall() throws UnknownHostException, HttpException {

    InetAddress[] expected = InetAddress.getAllByName(TEST_HOST);

    RoundRobinRouterPlaner routerPlaner = new RoundRobinRouterPlaner();
    HttpHost httpHost = new HttpHost(TEST_HOST);

    // first call
    HttpRoute firstRoute = routerPlaner.determineRoute(httpHost, null, null);

    for (int i = 0; i < expected.length; i++) {
        HttpRoute httpRouteNext = routerPlaner.determineRoute(httpHost, null, null);
        assertEquals(httpRouteNext.getTargetHost().getAddress(), firstRoute.getTargetHost().getAddress());
    }
}
 

private CloseableHttpClient httpClient() throws Exception {

        PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager(registry());

        Map<String, Object> httpClientMap = (Map<String, Object>)configMap.get(REST_TEMPLATE);
        connectionManager.setMaxTotal((Integer)httpClientMap.get(MAX_CONNECTION_TOTAL));
        connectionManager.setDefaultMaxPerRoute((Integer) httpClientMap.get(MAX_CONNECTION_PER_ROUTE));
        // Now handle all the specific route defined.
        Map<String, Object> routeMap = (Map<String, Object>)httpClientMap.get(ROUTES);
        Iterator<String> it = routeMap.keySet().iterator();
        while (it.hasNext()) {
            String route = it.next();
            Integer maxConnection = (Integer)routeMap.get(route);
            connectionManager.setMaxPerRoute(new HttpRoute(new HttpHost(
                    route)), maxConnection);
        }
        RequestConfig config = RequestConfig.custom()
                .setConnectTimeout((Integer)httpClientMap.get(TIMEOUT_MILLISECONDS))
                .build();

       return HttpClientBuilder.create()
                .setConnectionManager(connectionManager)
                .setDefaultRequestConfig(config).build();
    }
 

@Override public CloseableHttpResponse execute(HttpRoute route,
  org.apache.http.client.methods.HttpRequestWrapper request,
  HttpClientContext context, HttpExecutionAware execAware)
  throws IOException, HttpException {
  Span span = (Span) context.removeAttribute(Span.class.getName());

  if (span != null) {
    handler.handleSend(new HttpRequestWrapper(request, route.getTargetHost()), span);
  }

  CloseableHttpResponse response = mainExec.execute(route, request, context, execAware);
  if (span != null) {
    if (isRemote(context, span)) {
      if (serverName != null) span.remoteServiceName(serverName);
      parseTargetAddress(route.getTargetHost(), span);
    } else {
      span.kind(null); // clear as cache hit
    }
  }
  return response;
}
 

public static HttpClient generateHttpClient(final int maxTotalConnections,
                                            final int maxTotalConnectionsPerRoute, int connTimeout) {
  HttpParams params = new BasicHttpParams();
  ConnManagerParams.setMaxTotalConnections(params, maxTotalConnections);
  ConnManagerParams.setMaxConnectionsPerRoute(params, new ConnPerRoute() {
    @Override
    public int getMaxForRoute(HttpRoute route) {
      return maxTotalConnectionsPerRoute;
    }
  });
  HttpConnectionParams
      .setConnectionTimeout(params, connTimeout);
  SchemeRegistry schemeRegistry = new SchemeRegistry();
  schemeRegistry.register(
      new Scheme("http", PlainSocketFactory.getSocketFactory(), 80));
  SSLSocketFactory sslSocketFactory = SSLSocketFactory.getSocketFactory();
  sslSocketFactory.setHostnameVerifier(SSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
  schemeRegistry.register(new Scheme("https", sslSocketFactory, 443));
  ClientConnectionManager conMgr = new ThreadSafeClientConnManager(params, schemeRegistry);
  return new DefaultHttpClient(conMgr, params);
}
 

@Override
public CloseableHttpResponse execute(HttpRoute route, HttpRequestWrapper request,
    HttpClientContext context, HttpExecutionAware execAware) throws IOException, HttpException {
  if (request.getURI().getHost().endsWith(".mch.weixin.qq.com")) {
    return executeWithSignature(route, request, context, execAware);
  } else {
    return mainExec.execute(route, request, context, execAware);
  }
}
 

@Test
// Example 4.1
public final void whenIncreasingConnectionPool_thenNoEceptions() {
    poolingConnManager = new PoolingHttpClientConnectionManager();
    poolingConnManager.setMaxTotal(5);
    poolingConnManager.setDefaultMaxPerRoute(4);

    final HttpHost localhost = new HttpHost("locahost", 80);
    poolingConnManager.setMaxPerRoute(new HttpRoute(localhost), 5);
}
 

public static Map<HttpRoute, PoolStats> getConnManagerStats() {
    if (connManager != null) {
        Set<HttpRoute> routeSet = connManager.getRoutes();
        if (routeSet != null && !routeSet.isEmpty()) {
            Map<HttpRoute, PoolStats> routeStatsMap = new HashMap<HttpRoute, PoolStats>();
            for (HttpRoute route : routeSet) {
                PoolStats stats = connManager.getStats(route);
                routeStatsMap.put(route, stats);
            }
            return routeStatsMap;
        }
    }
    return null;
}
 

public static Map<HttpRoute, PoolStats> getConnManagerStats() {
    if (connManager != null) {
        Set<HttpRoute> routeSet = connManager.getRoutes();
        if (routeSet != null && !routeSet.isEmpty()) {
            Map<HttpRoute, PoolStats> routeStatsMap = new HashMap<HttpRoute, PoolStats>();
            for (HttpRoute route : routeSet) {
                PoolStats stats = connManager.getStats(route);
                routeStatsMap.put(route, stats);
            }
            return routeStatsMap;
        }
    }
    return null;
}
 

private HttpContext getContext( URI uri ) {
  HttpClientContext httpClientContext = HttpClientContext.create();
  //used by httpclient version >= 4.3
  httpClientContext
    .setAttribute( HttpClientContext.HTTP_ROUTE, new HttpRoute( new HttpHost( uri.getHost(), uri.getPort() ) ) );
  //used by httpclient version 4.2
  httpClientContext
    .setAttribute( HttpClientContext.HTTP_TARGET_HOST, new HttpHost( uri.getHost(), uri.getPort() ) );
  return httpClientContext;
}
 

@BeforeEach
@SuppressWarnings("unchecked")
void setup() {
    connPoolControl = (ConnPoolControl<HttpRoute>) mock(ConnPoolControl.class);
    binder = new PoolingHttpClientConnectionManagerMetricsBinder(connPoolControl, "test");
    binder.bindTo(registry);
}
 

@Test
// 7.1
public final void whenConfiguringTimeOut_thenNoExceptions() {
    route = new HttpRoute(new HttpHost("localhost", 80));
    poolingConnManager = new PoolingHttpClientConnectionManager();
    poolingConnManager.setSocketConfig(route.getTargetHost(), SocketConfig.custom().setSoTimeout(5000).build());
    assertTrue(poolingConnManager.getSocketConfig(route.getTargetHost()).getSoTimeout() == 5000);
}
 

@Before
public void setUp() throws Exception {
    reactor = new NioReactor();

    crusher = TcpCrusherBuilder.builder()
            .withReactor(reactor)
            .withBindAddress("127.0.0.1", CRUSHER_PORT)
            .withConnectAddress(REMOTE_HOST, REMOTE_PORT)
            .buildAndOpen();

    DnsResolver dnsResolver = new SystemDefaultDnsResolver() {
        @Override
        public InetAddress[] resolve(final String host) throws UnknownHostException {
            if (host.equalsIgnoreCase(REMOTE_HOST)) {
                return new InetAddress[] { InetAddress.getByAddress(new byte[] {127, 0, 0, 1}) };
            } else {
                return super.resolve(host);
            }
        }
    };

    HttpConnectionFactory<HttpRoute, ManagedHttpClientConnection> httpConnectionFactory =
            new ManagedHttpClientConnectionFactory();

    Registry<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder.<ConnectionSocketFactory>create()
            .register("http", PlainConnectionSocketFactory.INSTANCE)
            .build();

    PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager(
            socketFactoryRegistry, httpConnectionFactory, dnsResolver);

    http = HttpClients.createMinimal(connectionManager);
}