闲鱼智能客服架构演进：如何通过异步消息队列提升10倍处理效率-编程阁

背景：双11那2秒的“尴尬”

去年双11零点，闲鱼智能客服的 P99 延迟直接飙到 2.3 s，客服同学疯狂截图“转圈圈”。
根因很简单：同步 Servlet 线程池 + 下游 5 个 RPC 串行调用，只要有一个接口抖一下，整条链路就“堵车”。
大促峰值 4 k QPS，机器加到了 200 台，CPU 才 30%，线程却全部 Block 在 I/O 等待上——典型的“线程池打满但 CPU 闲得发慌”。
老板一句话：不解决 2 秒延迟，就把我的 OKR 打成 2 分。于是我们把目光投向了“异步消息队列”。

技术选型：Kafka 为什么能赢

先给出结论：Kafka 在顺序性、吞吐、运维成本三维打分最高，最终胜出。

维度	Kafka	RabbitMQ	RocketMQ
消息顺序性	分区级顺序，够用	队列级顺序，单队列吞吐低	分区级顺序，功能同 Kafka
单机吞吐	100 k+ QPS	3-5 k QPS	7-8 k QPS
运维成本	零 ZK（2.8+），机器少	镜像队列 + HA，节点多	专有 NameServer，额外组件
社区/生态	Spring 深度集成	老牌成熟	阿里内网资料多，但社区略小

客服场景只要“同一用户会话”有序即可，分区级顺序完全满足；再加上双11目标 5 k→50 k QPS，Kafka 的磁盘顺序写和零拷贝简直量身定做。
于是拍板：Kafka，3 台 16C32G 物理机，万兆网卡，成本不到前端机器加机器的 1/5。

核心实现：三步把同步改成异步

1. 请求-响应异步解耦

思路一句话：网关线程只负责“发消息 + 返回 ticket”，后端消费完再回调前端。

/** * 接收用户提问，发送 Kafka 后立刻返回 ticket * @param askDTO 用户问题 * @return 取票号，用于长轮询结果 */ @PostMapping("/ask") public ApiResult<String> ask(@ging AskDTO askDTO) { String ticket = IdUtil.fastUUID(); AskEvent event = AskEvent.builder() .ticket(ticket) .uid(askDTO.getUid()) .question(askDTO.getQuestion()) .timestamp(System.currentTimeMillis()) .build(); kafkaTemplate.send("ask-topic", askDTO.getUid(), event); return ApiResult.success(ticket); }

2. 批量消费 + 手动提交

Spring-kafka 的BatchListener一次拉 200 条，攒够 500 ms 或者 16 M 数据就处理，比单条消费吞吐高 8 倍。

@KafkaListener(topics = "ask-topic", groupId = "cs-group") public void batchAsk(List<ConsumerRecord<String, AskEvent>> records, Acknowledgment ack) { List<AskEvent> valid = records.stream() .filter(r -> !duplicateService.isDuplicate(r.value().getMsgId())) .map(r -> r.value()) .collect(toList()); if (valid.isEmpty()) { ack.acknowledge(); // 全部重复，直接提交 return; } // 1. 调用 NLP 模型 List<AnswerEvent> answers = nlpService.batchAsk(valid); // 2. 结果写回 Redis，供前端长轮询 answerPipeline.batchSave(answers); ack.acknowledge(); // 手动提交 offset }

3. 去重 & 死信

每条事件带全局 msgId（雪花算法），消费前用 Redis setnx 做幂等；失败 3 次进死信队列，人工兜底。

public boolean isDuplicate(String msgId) { return !redisTemplate.opsForValue() .setIfAbsent("dup:" + msgId, "1", Duration.ofMinutes(10)); }

死信处理器：

@KafkaListener(topics = "ask-topic.DLT", groupId = "dlt-group") public void handleDead(ConsumerRecord<String, AskEvent> r) { log.error("[DeadLetter] msgId={}", r.value().getMsgId()); // 发钉钉 + 落库，人工介入 }

性能测试：从 5 k 到 50 k 的跳跃

JMeter 200 线程压 10 min，数据对比如下：

指标	同步 Servlet	异步 Kafka
平均延迟	1200 ms	120 ms
P99 延迟	2300 ms	280 ms
峰值吞吐	5 k QPS	50 k QPS
CPU 利用率	30%	65%

并发调优公式：
分区数 = 目标吞吐 / 单线程最大吞吐 ≈ 50000 / 3000 ≈ 18
消费者并发 = 分区数 = 18
留 20% 余量，最终 24 分区，18 个 consumer 实例，每台 2 个线程，刚好打满网卡 70%，ISR 列表稳定。

避坑指南：Rebalance 与重复消费

避免 Rebalance
- session.timeout.ms=30s（默认 10 s 太短，GC 抖动就超时）
- max.poll.interval.ms=300s（批量处理慢任务必备）
- partition.assignment.strategy=CooperativeStickyAssignor（减少全局重平衡）
消息回溯导致重复
当 consumer 宕机重启，Kafka 根据“最后提交 offset”重放，可能重复。
解决：
- 业务侧幂等（本文已用 Redis setnx）
- 开启enable.idempotence=true+isolation.level=read_committed，避免事务消息重复
磁盘写满
日志段默认保留 7 天，双 11 流量大，磁盘 2 小时就涨 1 T。
动态调整：kafka-configs --alter --add-config retention.ms=86400000（改成 1 天），凌晨自动删除，白天稳如狗。