shop-starter-http-client：Spring Boot 3.5 微服务 HTTP 客户端基础设施设计

@Configuration(proxyBeanMethods = false)
class BuyerBffClientConfiguration {

    @Bean
    MarketplaceServiceClient marketplaceClient(
            ShopHttpExchangeSupport support,
            @Value("${shop.buyer.marketplace-service-url}") String url) {
        return support.createClient(url, MarketplaceServiceClient.class);
    }

    // 慢服务单独覆盖 read-timeout
    @Bean
    SearchServiceClient searchClient(
            ShopHttpExchangeSupport support,
            @Value("${shop.buyer.search-service-url}") String url) {
        return support.createClient(url, SearchServiceClient.class, Duration.ofSeconds(10));
    }
}

// buyer-bff/src/main/java/.../client/MarketplaceServiceClient.java
@HttpExchange
public interface MarketplaceServiceClient {
    @PostExchange(MarketplaceApi.LIST)
    ApiResponse<MarketplaceApi.ProductsView> listProducts(
            @RequestBody MarketplaceApi.ListProductsRequest request);
}

public class TracingHeaderInterceptor implements ClientHttpRequestInterceptor {

    private static final Set<String> DENIED_HEADERS =
            Set.of("Authorization", "Cookie", "Set-Cookie");

    @Override
    public ClientHttpResponse intercept(HttpRequest request, byte[] body,
                                        ClientHttpRequestExecution execution) throws IOException {
        if (tracer == null) return execution.execute(request, body);

        for (BaggageMapping mapping : baggageMappings) {
            if (DENIED_HEADERS.contains(mapping.headerName())) continue;
            String value = tracer.getBaggage(mapping.baggageName()).get();
            if (value != null && !value.isBlank()) {
                request.getHeaders().set(mapping.headerName(), value);
            }
        }
        return execution.execute(request, body);
    }
}

public void handleError(HttpRequest request, ClientHttpResponse response) {
    int status = response.getStatusCode().value();
    String code = CommonErrorCode.DOWNSTREAM_ERROR.getCode();
    String message = "Downstream request failed: " + request.getURI();

    try (var is = response.getBody()) {
        if (is != null) {
            byte[] body = is.readAllBytes();
            if (body.length > 0) {
                JsonNode json = objectMapper.readTree(body);
                String parsedCode = extractCode(json);
                String parsedMsg  = extractMessage(json);
                if (!parsedCode.isBlank()) code    = parsedCode;
                if (!parsedMsg.isBlank())  message = parsedMsg;
            }
        }
    } catch (IOException e) {
        log.warn("Failed to parse downstream error body [uri={}, status={}]",
                request.getURI(), status, e);
    }

    DownstreamException ex = new DownstreamException(status, code, message);
    enrichSpan(ex);
    throw ex;
}

private void enrichSpan(DownstreamException ex) {
    if (tracer == null) return;
    Span span = tracer.currentSpan();
    if (span == null) return;
    span.tag("downstream.http.status", String.valueOf(ex.getDownstreamHttpStatus()));
    span.tag("downstream.error.code", ex.getDownstreamCode());
    if (ex.getDownstreamHttpStatus() >= 500) span.error(ex);  // 4xx 不标 error
}

private static String extractCode(JsonNode body) {
    String code = body.path("status").asText();           // ApiResponse: SC_*
    if (!code.isBlank() && code.startsWith("SC_")) return code;

    code = body.path("code").asText();                    // 通用 code 字段
    if (!code.isBlank()) return code;

    String type = body.path("type").asText();             // ProblemDetail (RFC 7807)
    if (!type.isBlank() && type.contains("/"))
        return type.substring(type.lastIndexOf('/') + 1);

    return "";
}

public class DownstreamException extends BusinessException {
    private final int downstreamHttpStatus;  // 404 / 422 / 503 ...
    private final String downstreamCode;     // SC_NOT_FOUND / validation-failed ...
}

public class ShopHttpExchangeSupport {

    private static final ClientRequestObservationConvention OBSERVATION_CONVENTION =
            new DownstreamServiceObservationConvention();

    private final Supplier<RestClient.Builder> restClientBuilderSupplier;
    private final HttpClient httpClient;  // 共享连接池

    public <T> T createClient(String baseUrl, Class<T> clientClass, Duration readTimeout) {
        var requestFactory = new JdkClientHttpRequestFactory(httpClient);
        requestFactory.setReadTimeout(readTimeout);

        RestClient restClient = restClientBuilderSupplier.get()  // 每次新 builder
                .requestFactory(requestFactory)
                .baseUrl(baseUrl)
                .requestInterceptor(tracingInterceptor)
                .defaultStatusHandler(HttpStatusCode::isError, errorHandler::handleError)
                .observationConvention(OBSERVATION_CONVENTION)
                .build();

        return HttpServiceProxyFactory
                .builderFor(RestClientAdapter.create(restClient))
                .build()
                .createClient(clientClass);
    }
}

this.httpClient = HttpClient.newBuilder()
        .connectTimeout(connectTimeout != null ? connectTimeout : Duration.ofSeconds(2))
        .build();

public class DownstreamServiceObservationConvention
        extends DefaultClientRequestObservationConvention {

    @Override
    public KeyValues getLowCardinalityKeyValues(ClientRequestObservationContext ctx) {
        return super.getLowCardinalityKeyValues(ctx)
                .and(KeyValue.of("downstream.service", extractServiceName(ctx)));
    }

    private static String extractServiceName(ClientRequestObservationContext ctx) {
        try {
            ClientHttpRequest carrier = ctx.getCarrier();
            if (carrier != null) {
                String host = carrier.getURI().getHost();
                if (host != null && !host.isBlank()) return host;
            }
        } catch (Exception ignored) {}
        return "unknown";
    }
}

public BuyerApi.DashboardResponse loadDashboard(String buyerId) {
    ContextSnapshot snapshot = ContextSnapshot.captureAll();
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        var profile     = executor.submit(snapshot.wrap(() -> getProfile(buyerId)));
        var wallet      = executor.submit(snapshot.wrap(() -> getWallet(buyerId)));
        var marketplace = executor.submit(snapshot.wrap(() -> listMarketplace()));
        var loyalty     = executor.submit(snapshot.wrap(() -> getLoyaltyAccount(buyerId)));
        // ...
    }
}

management:
  tracing:
    baggage:
      remote-fields: [X-Buyer-Id, X-Seller-Id, X-Username, X-Portal, X-Roles, X-Order-Id, X-Request-Id]
      correlation-fields: [X-Buyer-Id, X-Seller-Id, X-Username, X-Order-Id]

shop:
  http-client:
    connect-timeout: 2s
    read-timeout: 5s
    baggage-headers: []   # 空则使用 BaggageMapping.defaults() 的 6 个 header

management:
  observations:
    http:
      server:
        requests:
          write-trace-header: true

# 1. 指定服务的错误率
sum(rate(http_client_requests_seconds_count{
    downstream_service="marketplace-service",
    outcome="SERVER_ERROR"}[5m]))
/
sum(rate(http_client_requests_seconds_count{
    downstream_service="marketplace-service"}[5m]))

# 2. 所有服务并排比较（每个服务一条线）
sum by (downstream_service) (rate(http_client_requests_seconds_count{
    outcome=~"SERVER_ERROR|CLIENT_ERROR"}[5m]))
/
sum by (downstream_service) (rate(http_client_requests_seconds_count[5m]))

# 3. 每个服务的 P99 延迟
histogram_quantile(0.99,
  sum by (downstream_service, le) (
    rate(http_client_requests_seconds_bucket[5m])))

# 4. 总 RPS 排名（识别热点服务）
topk(5, sum by (downstream_service) (
    rate(http_client_requests_seconds_count[5m])))

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

management:
  endpoints:
    web:
      exposure:
        include: prometheus, health
  server:
    port: 8081     # 应用端口 8080，管理端口 8081 隔离

$ curl -s http://buyer-bff:8081/actuator/prometheus | grep http_client_requests_seconds_count | head -2
http_client_requests_seconds_count{downstream_service="marketplace-service",method="GET",outcome="SUCCESS",status="200",uri="/api/marketplace/products",} 142.0
http_client_requests_seconds_count{downstream_service="loyalty-service",method="GET",outcome="SUCCESS",status="200",uri="/api/loyalty/account/{buyerId}",} 98.0

groups:
  - name: http-client
    rules:
      - alert: DownstreamHighErrorRate
        expr: |
          sum by (downstream_service) (
            rate(http_client_requests_seconds_count{outcome=~"SERVER_ERROR|CLIENT_ERROR"}[5m]))
          / sum by (downstream_service) (
            rate(http_client_requests_seconds_count[5m]))
          > 0.01          
        for: 5m
        labels:    { severity: warning }
        annotations:
          summary: "下游 {{ $labels.downstream_service }} 错误率 > 1%（5m）"

      - alert: DownstreamHighLatencyP99
        expr: |
          histogram_quantile(0.99,
            sum by (downstream_service, le) (
              rate(http_client_requests_seconds_bucket[5m]))) > 0.5          
        for: 5m
        labels:    { severity: warning }
        annotations:
          summary: "下游 {{ $labels.downstream_service }} P99 > 500ms（5m）"

      - alert: CircuitBreakerOpen
        expr: resilience4j_circuitbreaker_state{state="open"} > 0
        for: 1m
        labels:    { severity: critical }
        annotations:
          summary: "熔断器 {{ $labels.name }} 已开路"

logging:
  level:
    org.springframework.web.client.RestClient: DEBUG

shop:
  http-client:
    connect-timeout: 2s
    read-timeout: 5s

support.createClient(searchUrl, SearchServiceClient.class, Duration.ofSeconds(10));

@CircuitBreaker(name = "loyalty-service", fallbackMethod = "getLoyaltyAccountFallback")
public LoyaltyApi.AccountResponse getLoyaltyAccount(String buyerId) {
    return loyaltyServiceClient.getAccount(buyerId).data();
}

resilience4j:
  circuitbreaker:
    instances:
      loyalty-service:
        sliding-window-size: 10
        failure-rate-threshold: 50           # 50% 失败率触发熔断
        wait-duration-in-open-state: 30s     # 熔断后 30s 再尝试半开
        permitted-number-of-calls-in-half-open-state: 3
        ignore-exceptions:
          - dev.meirong.shop.common.error.DownstreamException  # 4xx 业务拒绝不计失败

private LoyaltyApi.AccountResponse getLoyaltyAccountFallback(String buyerId, Throwable t) {
    log.warn("loyalty-service unavailable [buyerId={}]", buyerId);
    return LoyaltyApi.AccountResponse.empty();
}

@Retry(name = "search-service")
@GetExchange(SearchApi.SEARCH)
ApiResponse<SearchApi.SearchResult> search(@RequestParam String q);

resilience4j:
  retry:
    instances:
      search-service:
        max-attempts: 2
        wait-duration: 100ms
        retry-exceptions:
          - java.net.ConnectException
          - java.net.SocketTimeoutException
        ignore-exceptions:
          - dev.meirong.shop.common.error.DownstreamException  # 4xx/5xx 一律不重试