Agent API 架构设计与性能优化实战：从高并发瓶颈到弹性伸缩解决方案

1次阅读

共计 1965 个字符，预计需要花费 5 分钟才能阅读完成。

最近在重构公司内部的 Agent API 服务时，遇到了一个典型的高并发问题。原有的同步阻塞式 API 在面对突发流量时，出现了严重的雪崩效应。从 Nginx 的日志统计图中可以清晰地看到，当 QPS 超过 2000 时，响应时间呈现指数级增长（见图 1）。

主要问题集中在：

数据库连接池被耗尽
线程阻塞导致 CPU 利用率居高不下
级联故障波及上下游服务

我们对比了三种常见的解决方案：

REST 长轮询
优点：实现简单，兼容性好
缺点：无效请求多，资源利用率低
WebSocket
优点：实时性高，连接复用
缺点：连接维护成本高，不适合短连接场景
消息队列（Kafka）
优点：高吞吐量，削峰填谷
缺点：系统复杂度增加

最终选择了基于 Kafka 的异步消息方案，实测吞吐量提升了 8 倍（从 2k QPS 到 16k QPS）。

使用 Spring WebFlux 构建反应式 API 端点：

@RestController
@RequestMapping("/agent")
public class AgentController {@PostMapping("/event")
    public Mono<ResponseEntity<String>> handleEvent(@RequestBody EventRequest request) {return eventService.process(request)
            .onBackpressureBuffer(1000) // 背压处理
            .map(result -> ResponseEntity.ok(result));
    }
}

确保相同 Agent ID 的消息路由到同一分区：

public class AgentPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, 
        byte[] keyBytes, Object value, 
        byte[] valueBytes, Cluster cluster) {

        // 按 Agent ID 哈希取模
        return Math.abs(key.toString().hashCode()) % cluster.partitionCountForTopic(topic);
    }
}

批量查询 Agent 状态：

StatefulRedisConnection<String, String> connection = redisClient.connect();
RedisAsyncCommands<String, String> commands = connection.async();

commands.setAutoFlushCommands(false);

List<RedisFuture<String>> futures = new ArrayList<>();
for (String agentId : agentIds) {futures.add(commands.get("agent:state:" + agentId));
}

commands.flushCommands();

// 处理所有结果
List<String> states = LettuceFutures.awaitAll(10, TimeUnit.SECONDS, 
    futures.toArray(new RedisFuture[0]));

测试环境配置：
– 4 台 16 核 32G 的 K8s Pod
– Kafka 3 节点集群
– Redis Cluster 6 节点

压测对比数据：

指标	优化前	优化后
最大 QPS	2,000	16,000
TP99 延迟 (ms)	1,200	85
错误率	8.7%	0.01%

分布式锁问题
使用 Redisson 的看门狗机制避免锁过期
为自动扩缩容场景实现锁续期逻辑

消息积压处理

# 动态调整消费者数量示例
def adjust_consumers():
    lag = get_kafka_lag()
    if lag > 10000:
        scale_consumers(+2)
    elif lag < 1000:
        scale_consumers(-1)