---

背景痛点：流量一涨，客服先“宕”

去年双十一，我们内部一套老客服系统直接“罢工”——QPS 从日常 300 飙到 1800，CPU 打满，RT 从 200 ms 涨到 3 s，用户疯狂点“转人工”，结果人工坐席也进不来。
事后复盘，瓶颈集中在三点：

1. 每个对话长连接都占用一个线程，4C8G 机器上线程数飙到 1.2 w，内核调度直接炸锅。
2. NLU 模型推理是同步的，一条“我要退货”要跑 80 ms，线程被卡住，请求排队。
3. 微服务之间每次 RPC 都新建 TCP 连接，TIME_WAIT 把端口耗尽，只能重启 Pod 续命。

一句话：同步 + 阻塞 + 无复用，高并发面前就是“三连跪”。

架构对比：同步线程模型 vs Reactor+协程池

老架构（SpringBoot+Tomcat）典型同步流程：

请求→Tomcat 线程→NLU 同步调用→DB 查询→返回

线程数≈并发数，等待 IO 时线程空转，内存白白浪费。

蜂答思路：把“等”变成“回调”。

• 网络层：epoll 单线程 Reactor，收到请求后封装成 Task，扔进 GMP 调度器。
• 计算层：Goroutine Pool 只开 GOMAXPROCS*2 条工作协程，通过 Channel 消费任务；NLU、DB、RPC 全部用 context.WithTimeout 做异步 IO。
• 存储层：Kafka 解耦，答题结果先落内存队列，批量写 MySQL，降低 70% IOPS。

结果同样 4C8G，老架构 500 QPS 就 90% CPU，蜂答可以稳在 5000+ QPS，CPU 只到 55%，内存反而降了 30%。

核心实现：连接池+异步消息

1. Golang 连接池（带自动回收）

下面代码基于 github.com/jolestar/go-commons-pool/v2 二次封装，重点在 Eviction 定时清理空闲连接，防止 MySQL 端“Too many connections”。

package pool

import (
	"context"
	"database/sql"
	"time"

	"github.com/jolestar/go-commons-pool/v2"
	_ "github.com/go-sql-driver/mysql"
)

type connFactory struct {
	dsn string
}

func (f *connFactory) MakeObject(ctx context.Context) (*pool.PooledObject, error) {
	db, err := sql.Open("mysql", f.dsn)
	if err != nil {
		return nil, err
	}
	// 真正探活
	if err = db.PingContext(ctx); err != nil {
		return nil, err
	}
	return pool.NewPooledObject(db), nil
}

func (f *connFactory) DestroyObject(ctx context.Context, obj *pool.PooledObject) error {
	return obj.Object.(*sql.DB).Close()
}

// 空闲超 30 s 就干掉，防止 DB 端堆积
func (f *connFactory) EvictionConfig() *pool.BaseEvictionConfig {
	return &pool.BaseEvictionConfig{
		MinEvictableIdleTime: 30 * time.Second,
		TimeBetweenEviction:  10 * time.Second,
	}
}

func NewConnPool(dsn string, maxIdle, maxActive int) *pool.ObjectPool {
	f := &connFactory{dsn: dsn}
	config := pool.NewDefaultPoolConfig()
	config.MaxTotal = maxActive
	config.MaxIdle = maxIdle
	return pool.NewObjectPool(context.Background(), f, config)
}

使用示例：

p := NewConnPool(dsn, 20, 200)
v, _ := p.BorrowObject(context.Background())
db := v.(*sql.DB)
// 业务 SQL
p.ReturnObject(context.Background(), v)

2. Kafka 异步消息流程

1. API 网关把用户提问写进 Kafka question-in 分区，Key=UserID，保序。
2. NLU-Consumer 组（可水平扩容）消费，计算意图，结果写 answer-out。
3. Websocket-Gateway 监听 answer-out，通过本地内存 map 查到对应长连接，Push 回前端。

全程无共享 DB 锁，只靠 Kafka 的 offset 保证“至少一次”，业务层做幂等（UUID 去重表）。

性能验证：从 500 到 5000 QPS 的硬数据

1. 压测脚本（wrk）

wrk -t16 -c1000 -d60s --latency -s query.lua https://gateway.fengda.com/api/ask

query.lua 内容：

wrk.method = "POST"
wrk.body   = '{"q":"我要退货","uid":'..math.random(1,1000000)..'}'
wrk.headers["Content-Type"] = "application/json"

2. 监控对比

指标	同步架构(500 QPS)	蜂答(5000 QPS)
CPU 使用率	92%	55%
内存占用	5.8 G	4.1 G
平均 RT	220 ms	38 ms
P99 RT	1.8 s	120 ms
TCP 新建/s	1.1 w	800
GC 次数/60 s	2.3 k	180

GC 次数大幅下降，得益于对象复用 + 内存池，STW 时间从 30 ms 降到 3 ms，长尾请求几乎消失。

避坑指南：锁、并发、安全

1. 分布式锁别乱用

会话保持场景，很多同学习惯用 Redis 分布式锁占坑：

//  错误示范
if redis.SetNX(ctx, key, 1, 30*time.Second) {
    // 处理消息
}

高并发下，SetNX 成功≠你处理完，GC 或网络抖动导致锁过期，另一个协程重复消费，用户收到两条答案。
正确姿势：锁+本地队列双保险。

//  推荐
type Slot struct {
	ch   chan Message
	last int64 // 上次收到消息时间
}

slots := make(map[string]*Slot)
mu   sync.RWMutex

// 同一个 UserID 永远进同一个 chan
mu.RLock()
s, ok := slots[uid]
mu.RUnlock()
if !ok {
	mu.Lock()
	s = &Slot{ch: make(chan Message, 128)}
	slots[uid] = s
	mu.Unlock()
	go consume(uid, s.ch) // 单协程顺序消费
}
s.ch <- msg

2. 敏感词过滤的线程安全

DFA 词库初始化后只读，但命中记录要实时写缓存。
如果直接用 map[string]int 累加，并发写会 panic。
用 sync.Map 或 atomic.AddInt64 都行，推荐拆成 shard 分片，减少伪共享：

type HitCounter [256]atomic.Int64

func (h *HitCounter) Add(word string) {
	idx := fnv32(word) & 255
	h[idx].Add(1)
}

延伸思考：动态扩缩容怎么玩？

Kafka 方案天然带背压，只要保证 Consumer Lag 可控，就能按 Lag 扩缩容。
我们做了三层指标：

1. Lag > 3 w 且持续 30 s → HPA 触发，NLU Pod 副本数+50%。
2. CPU < 20% 且 Lag < 1 w 持续 5 min → 副本数-25%，最低保留 2 实例。
3. 分区重分配：扩容后调用 Kafka Admin API，把分区平均到新 Broker，避免热点。

配合 K8s 的 hpa/v2，整个流程全自动，去年双十一高峰 2 min 内把 20 个 Pod 拉到 120 个，流量回落后又缩回 20，省下的机器直接给算法同学跑模型，老板直呼“省钱小能手”。

---

整套代码已经跑在我们 GitLab 私有库，如果你也在为客服系统并发发愁，不妨把连接池和 Kafka 解耦思路先搬过去跑一跑，改两行配置就能让 QPS 翻几番。
调优路上，欢迎一起踩坑交流。

---