feat: optimize http long link

Author: wardseptember

trpc-proto/trpc-proto-http/src/main/java/com/tencent/trpc/proto/http/client/Http2ConsumerInvoker.java

@@ -44,11 +44,6 @@
 
 /**
  * HTTP/2 protocol client invoker, supporting both h2 and http2c.
- *
- * <p>Each {@link #send(Request)} entry signals the underlying {@link Http2cRpcClient} that
- * it is being used (drives the idle-eviction heuristic) and reports success / failure to
- * drive the consecutive-failure counter that flips the client to unavailable on sustained
- * backend outages.</p>
  */
 public class Http2ConsumerInvoker<T> extends AbstractConsumerInvoker<T> {
 
@@ -72,31 +67,20 @@ public Http2ConsumerInvoker(Http2cRpcClient client, ConsumerConfig<T> config,
     @Override
     public Response send(Request request) throws Exception {
         Http2cRpcClient http2cRpcClient = (Http2cRpcClient) client;
-        // Mark "used" before any work so even a failed request keeps the idle-eviction timer
-        // accurate (a failing client is still actively used and must not be reaped as orphan).
-        http2cRpcClient.markUsed();
 
         int requestTimeout = config.getBackendConfig().getRequestTimeout();
         SimpleHttpRequest simpleHttpRequest;
         try {
             simpleHttpRequest = buildRequest(request, requestTimeout);
         } catch (Exception ex) {
-            http2cRpcClient.markFailure();
             return RpcUtils.newResponse(request, null, ex);
         }
 
         try {
             SimpleHttpResponse simpleHttpResponse = execute(request, requestTimeout,
                     simpleHttpRequest, http2cRpcClient);
-            Response response = handleResponse(request, simpleHttpResponse);
-            if (response.getException() == null) {
-                http2cRpcClient.markSuccess();
-            } else {
-                http2cRpcClient.markFailure();
-            }
-            return response;
+            return handleResponse(request, simpleHttpResponse);
         } catch (Exception e) {
-            http2cRpcClient.markFailure();
             return RpcUtils.newResponse(request, null, e);
         }
 
@@ -164,42 +148,56 @@ private SimpleHttpResponse execute(Request request, int requestTimeout,
             SimpleHttpRequest simpleHttpRequest, Http2cRpcClient http2cRpcClient) throws Exception {
         CloseableHttpAsyncClient httpAsyncClient = http2cRpcClient.getHttpAsyncClient();
         Future<SimpleHttpResponse> httpResponseFuture = httpAsyncClient.execute(simpleHttpRequest,
-                new FutureCallback<SimpleHttpResponse>() {
-                    @Override
-                    public void completed(SimpleHttpResponse result) {
-                        if (logger.isDebugEnabled()) {
-                            logger.debug(result.getBodyText());
-                        }
-                    }
-
-                    @Override
-                    public void failed(Exception ex) {
-                        String msg = String
-                                .format("request has exception > %s ms, service=%s, "
-                                                + "method=%s, remoteAddr=%s, exception=%s",
-                                        requestTimeout,
-                                        request.getInvocation().getRpcServiceName(),
-                                        request.getInvocation().getRpcMethodName(),
-                                        request.getMeta().getRemoteAddress(),
-                                        ex.getMessage());
-                        logger.error(msg);
-                    }
-
-                    @Override
-                    public void cancelled() {
-                        String msg = String
-                                .format("request cancel > %s ms, service=%s, "
-                                                + "method=%s, remoteAddr=%s",
-                                        requestTimeout,
-                                        request.getInvocation().getRpcServiceName(),
-                                        request.getInvocation().getRpcMethodName(),
-                                        request.getMeta().getRemoteAddress());
-                        logger.error(msg);
-                    }
-                });
+                newResponseCallback(request, requestTimeout));
         return httpResponseFuture.get(requestTimeout, TimeUnit.MILLISECONDS);
     }
 
+    /**
+     * Build the {@link FutureCallback} used to log the asynchronous request outcome. Extracted
+     * (package-private) from {@link #execute} so the completed / failed / cancelled logging
+     * branches can be unit-tested directly — the convenience {@code execute(SimpleHttpRequest,
+     * FutureCallback)} on {@link CloseableHttpAsyncClient} is {@code final} and cannot be stubbed.
+     *
+     * @param request the originating TRPC request (used only for log context)
+     * @param requestTimeout the request timeout in milliseconds (used only for log context)
+     * @return a logging-only callback; it never mutates request/response state
+     */
+    FutureCallback<SimpleHttpResponse> newResponseCallback(Request request, int requestTimeout) {
+        return new FutureCallback<SimpleHttpResponse>() {
+            @Override
+            public void completed(SimpleHttpResponse result) {
+                if (logger.isDebugEnabled()) {
+                    logger.debug(result.getBodyText());
+                }
+            }
+
+            @Override
+            public void failed(Exception ex) {
+                String msg = String
+                        .format("request has exception > %s ms, service=%s, "
+                                        + "method=%s, remoteAddr=%s, exception=%s",
+                                requestTimeout,
+                                request.getInvocation().getRpcServiceName(),
+                                request.getInvocation().getRpcMethodName(),
+                                request.getMeta().getRemoteAddress(),
+                                ex.getMessage());
+                logger.error(msg);
+            }
+
+            @Override
+            public void cancelled() {
+                String msg = String
+                        .format("request cancel > %s ms, service=%s, "
+                                        + "method=%s, remoteAddr=%s",
+                                requestTimeout,
+                                request.getInvocation().getRpcServiceName(),
+                                request.getInvocation().getRpcMethodName(),
+                                request.getMeta().getRemoteAddress());
+                logger.error(msg);
+            }
+        };
+    }
+
     /**
      * Wrap the TRPC request into an HTTP request.
      *

trpc-proto/trpc-proto-http/src/main/java/com/tencent/trpc/proto/http/client/Http2RpcClient.java

@@ -11,6 +11,7 @@
 
 package com.tencent.trpc.proto.http.client;
 
+import static com.tencent.trpc.proto.http.common.HttpConstants.VALIDATE_AFTER_INACTIVITY_MS;
 import static com.tencent.trpc.transport.http.common.Constants.KEYSTORE_PASS;
 import static com.tencent.trpc.transport.http.common.Constants.KEYSTORE_PATH;
 
@@ -21,6 +22,7 @@
 import com.tencent.trpc.core.logger.LoggerFactory;
 import java.io.File;
 import javax.net.ssl.SSLContext;
+import org.apache.hc.client5.http.impl.async.HttpAsyncClientBuilder;
 import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
 import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManager;
 import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManagerBuilder;
@@ -32,23 +34,17 @@
 import org.apache.hc.core5.util.TimeValue;
 
 /**
- * HTTP/2 (TLS) protocol client. Inherits long-connection state ({@code lastUsedNanos},
- * {@code consecutiveFailures}, {@link #markUsed}, {@link #markSuccess}, {@link #markFailure}
- * and the overridden {@link #isAvailable()}) from {@link Http2cRpcClient}; the differences
- * are the TLS handshake and the explicit {@link HttpVersionPolicy} negotiation.
+ * HTTP/2 (TLS) protocol client. Inherits the long-connection connection-pool setup from
+ * {@link Http2cRpcClient}; the differences are the TLS handshake and the explicit
+ * {@link HttpVersionPolicy} negotiation.
  *
  * <p>The connection manager is sized and tuned identically to {@link Http2cRpcClient}: pool
- * limits derived from {@code maxConns}, idle / expired eviction, SO_KEEPALIVE and a hard
- * connection TTL.</p>
+ * limits derived from {@code maxConns}, idle / expired eviction and SO_KEEPALIVE.</p>
  */
 public class Http2RpcClient extends Http2cRpcClient {
 
     private static final Logger logger = LoggerFactory.getLogger(Http2RpcClient.class);
 
-    private static final int VALIDATE_AFTER_INACTIVITY_MS = 2000;
-    private static final long EVICT_IDLE_CONNECTIONS_SECONDS = 60L;
-    private static final int CONNECTION_TTL_MINUTES = 10;
-
     /**
      * The protocol type used for interaction with the server, such as HTTP1, H2, or protocol negotiation.
      * In trpc, the interaction is forced to use H2 or HTTP1 protocol based on the configuration.
@@ -84,20 +80,19 @@ protected void doOpen() throws TRpcException {
                     .setMaxConnPerRoute(maxConns)
                     .setConnPoolPolicy(PoolReusePolicy.LIFO)
                     .setValidateAfterInactivity(TimeValue.ofMilliseconds(VALIDATE_AFTER_INACTIVITY_MS))
-                    .setConnectionTimeToLive(TimeValue.ofMinutes(CONNECTION_TTL_MINUTES))
                     .build();
 
             // 3. Configure the client to force HTTPS protocol to use HTTP1 communication and H2 protocol
             // to use H2 communication.
-            httpAsyncClient = HttpAsyncClients.custom()
+            HttpAsyncClientBuilder builder = HttpAsyncClients.custom()
                     .setVersionPolicy(this.clientVersionPolicy)
                     .setConnectionManager(cm)
                     .setIOReactorConfig(IOReactorConfig.custom()
                             .setSoKeepAlive(true)
                             .build())
-                    .evictExpiredConnections()
-                    .evictIdleConnections(TimeValue.ofSeconds(EVICT_IDLE_CONNECTIONS_SECONDS))
-                    .build();
+                    .evictExpiredConnections();
+            applyIdleEviction(builder);
+            httpAsyncClient = builder.build();
             // 4. Start the client.
             httpAsyncClient.start();
         } catch (Exception e) {