微服务链路怎样同时看响应与资源

接入模型服务后，响应时间和资源消耗应放在同一张观测表里。本文讨论缓存、并发、调用量与资源占用的取舍；代码中的参数只用于说明接口形态，不能直接当作生产配置。

模型调用会同时带来等待时间和用量支出，但这不意味着所有链路都需要批量聚合或语义缓存。先确认请求是否可合并、缓存结果是否可复用、降级结果是否可接受，再选择对应手段。

1. 建立可复查的诊断证据

排查上游服务调用链时，应把网关、调用方和模型服务的指标按同一请求关联起来。

通过 Prometheus 接口与网关日志抓取性能数据：

# 查询 AI 预测服务在 Resilience4j 中的调用延迟分布 (PromQL)
curl -g "${PROMETHEUS_URL}/api/v1/query?query=<url-encoded-promql>"

# 检查当前微服务节点上本地 Caffeine 缓存命中率与 eviction 计数
curl -s http://localhost:8080/actuator/metrics/caffeine.cache.hit.total

# 查看 API 网关层 Outbound HTTP 连接池状态与 Pending 队列
curl -s http://localhost:8080/actuator/prometheus | grep "reactor_netty_connection_provider"

下面的指标格式仅演示应如何取数，实际数值应来自当前环境：

# PromQL 指标展现：AI 预测微服务 P99 延迟高达 3.8 秒
resilience4j_circuitbreaker_calls_seconds_bucket{name="aiPredictService",le="3.8"} 8941

# Caffeine 缓存命中率：只有可怜的 4.2%
caffeine_cache_hit_total{name="semanticCache"} 421.0

# Netty 连接池 Pending 队列严重积压
reactor_netty_connection_provider_pending_connections{id="ai-provider-pool"} 380.0

深入代码发现：

1. 调用粒度不匹配：如果每个细小事件都单独请求模型，需要先确认它们是否真的不能合并。
2. 缓存键与复用目标不一致：精确文本键适合完全相同的输入；相近内容是否复用结果，需要业务校验与失效策略。
3. 请求缺少背压处理：当可合并请求持续积压时，才评估窗口聚合与队列上限，并保留超时和拒绝路径。

2. 微服务两级语义缓存与 Batching 批处理架构

为了兼顾系统 P99 延迟与 API 调用成本，我们在 Spring Cloud Gateway 与下游 AI 预测服务之间设计了两级缓存与动态 Batch 聚合防护架构。

三层防护机制：

1. L1/L2 两级缓存：L1 内存缓存拦截高频重复请求；L2 基于 Embedding 向量计算余弦相似度，相似度 > 0.95 的直接命中缓存，无需重复调用 LLM。
2. 响应式请求聚合：可使用 bufferTimeout 聚合可兼容的请求；窗口大小、队列上限和失败处理必须由压测与业务时限共同确定。
3. 弹性断路器兜底：当外部 API 延迟突破 2 秒或触发 429 限流时，Resilience4j 自动开启熔断，微服务无缝降级至本地规则引擎（Rule-based Engine）。

3. 生产级 Batch 聚合器与两级缓存降级代码

以下为 Spring Cloud 环境下基于 WebClient 与 Reactive Reactor 实现的自动化 Batch 聚合器：

package com.architecture.spcloud.ai.batch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

import java.time.Duration;
import java.util.List;
import java.util.Map;

@Service
public class AiPredictBatchAggregator {

    private static final Logger log = LoggerFactory.getLogger(AiPredictBatchAggregator.class);
    
    // 使用 Single-Producer Multi-Consumer Sink 接收高并发微小请求
    private final Sinks.Many<PredictTask> taskSink = Sinks.many().unicast().onBackpressureBuffer();
    private final WebClient webClient;

    public AiPredictBatchAggregator(WebClient.Builder webClientBuilder) {
        this.webClient = webClientBuilder.baseUrl("https://ai-provider.internal").build();
        initBatchProcessor();
    }

    public Mono<String> submitPrediction(String userId, String featureText) {
        PredictTask task = new PredictTask(userId, featureText);
        taskSink.tryEmitNext(task);
        return task.getResultMono();
    }

    private void initBatchProcessor() {
        taskSink.asFlux()
                // 100ms 窗口或凑齐 20 个请求即触发一次 Batch 发送
                .bufferTimeout(20, Duration.ofMillis(100))
                .flatMap(this::executeBatchRpc)
                .subscribe();
    }

    private Mono<Void> executeBatchRpc(List<PredictTask> batch) {
        if (batch.isEmpty()) return Mono.empty();

        log.info("触发 Batch 聚合 RPC 调用，包含请求数量: {}", batch.size());
        
        List<String> payloadList = batch.stream().map(PredictTask::getFeatureText).toList();

        return webClient.post()
                .uri("/v1/batch-predict")
                .bodyValue(Map.of("inputs", payloadList))
                .retrieve()
                .bodyToMono(BatchResponse.class)
                .doOnNext(response -> {
                    List<String> results = response.getResults();
                    for (int i = 0; i < batch.size(); i++) {
                        // 将结果写回对应的 Mono
                        batch.get(i).getResultSink().tryEmitValue(results.get(i));
                    }
                })
                .doOnError(throwable -> {
                    log.error("Batch RPC 执行失败，触发降级保护", throwable);
                    batch.forEach(task -> 
                        task.getResultSink().tryEmitValue("{\"status\": \"UNKNOWN\", \"fallback\": true}")
                    );
                })
                .then();
    }

    // 内部任务封装对象
    public static class PredictTask {
        private final String userId;
        private final String featureText;
        private final Sinks.One<String> resultSink = Sinks.one();

        public PredictTask(String userId, String featureText) {
            this.userId = userId;
            this.featureText = featureText;
        }

        public String getFeatureText() { return featureText; }
        public Mono<String> getResultMono() { return resultSink.asMono(); }
        public Sinks.One<String> getResultSink() { return resultSink; }
    }

    public static class BatchResponse {
        private List<String> results;
        public List<String> getResults() { return results; }
        public void setResults(List<String> results) { this.results = results; }
    }
}

针对成本控制的 L2 语义缓存与 Resilience4j 降级组件：

package com.architecture.spcloud.ai.cache;

import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.time.Duration;

@Service
public class SemanticCacheService {

    private final StringRedisTemplate redisTemplate;
    private final AiPredictBatchAggregator batchAggregator;

    public SemanticCacheService(StringRedisTemplate redisTemplate, AiPredictBatchAggregator batchAggregator) {
        this.redisTemplate = redisTemplate;
        this.batchAggregator = batchAggregator;
    }

    @CircuitBreaker(name = "aiPredictService", fallbackMethod = "fallbackPredict")
    public String getPredictionWithCache(String userId, String text) {
        String cacheKey = "ai:cache:" + computeSemanticHash(text);
        
        // 1. 查询 L2 语义缓存
        String cachedValue = redisTemplate.opsForValue().get(cacheKey);
        if (cachedValue != null) {
            return cachedValue;
        }

        // 2. 缓存未命中，提交至 Batch 聚合器执行 RPC
        String newValue = batchAggregator.submitPrediction(userId, text).block(Duration.ofSeconds(3));

        // 3. 异步写入 Redis 缓存，TTL 24 小时
        if (newValue != null && !newValue.contains("fallback")) {
            redisTemplate.opsForValue().set(cacheKey, newValue, Duration.ofHours(24));
        }

        return newValue;
    }

    public String fallbackPredict(String userId, String text, Throwable t) {
        // 熔断降级兜底逻辑：返回基于传统规则引擎的静态预测结果
        return "{\"status\": \"SAFE\", \"score\": 0.0, \"source\": \"RULE_ENGINE_FALLBACK\"}";
    }

    private String computeSemanticHash(String text) {
        // 真实生产环境可替换为向量相似度近邻计算 (如 HNSW 索引查询)
        return Integer.toHexString(text.trim().toLowerCase().hashCode());
    }
}

4. 优化后的验证口径

验证时至少记录四类数据：端到端延迟分位、调用量与缓存命中、队列拒绝或超时次数，以及降级结果的业务正确性。把优化前后在相同样本和相同负载下的结果并列保存；当缓存命中或批量处理影响结果一致性时，应以正确性优先。

微服务接入模型服务时，缓存、请求聚合和熔断都是可选手段。先定义可接受的等待时间、失败结果和资源预算，再用观测数据决定是否启用它们。