高吞吐场景下大模型推理服务的流式并发控制设计前言最近在做一个高吞吐的大模型推理服务要求支持 1000 QPS 的流式输出。一开始采用简单的 goroutine 并发模式结果遇到了严重的问题并发量上去后GPU 显存瞬间打满流式响应出现乱序服务在高负载下不稳定经过几轮优化最终实现了稳定的高吞吐流式服务。这篇文章分享完整的设计方案。一、 问题分析1.1 传统方案的瓶颈graph TD A[并发请求] -- B[goroutine] B -- C[GPU推理] C -- D[流式响应] E[无限制并发] --|导致 | F[显存爆炸] B -.- E1.2 性能数据指标传统方案问题峰值并发无限制显存 OOM响应顺序乱序客户端难以处理资源利用率不稳定GPU 利用率波动大服务可用性95%高负载下崩溃二、 解决方案流式并发控制器2.1 架构设计graph TD A[客户端请求] -- B[请求队列] B -- C[并发控制器] C -- D[GPU推理池] D -- E[响应缓冲] E -- F[流式输出] G[配置中心] -- C H[监控系统] -- C2.2 核心实现type StreamController struct { maxConcurrency int currentConcurrency int32 requestQueue chan *InferenceRequest responseQueue chan *StreamResponse wg sync.WaitGroup ctx context.Context cancel context.CancelFunc } func NewStreamController(maxConcurrency int) *StreamController { ctx, cancel : context.WithCancel(context.Background()) return StreamController{ maxConcurrency: maxConcurrency, requestQueue: make(chan *InferenceRequest, 10000), responseQueue: make(chan *StreamResponse, 10000), ctx: ctx, cancel: cancel, } } func (sc *StreamController) Start() { // 启动响应处理器 go sc.processResponses() // 启动工作协程 for i : 0; i sc.maxConcurrency; i { sc.wg.Add(1) go sc.worker() } } func (sc *StreamController) worker() { defer sc.wg.Done() for { select { case req : -sc.requestQueue: sc.processRequest(req) case -sc.ctx.Done(): return } } } func (sc *StreamController) processRequest(req *InferenceRequest) { // 原子增加并发计数 atomic.AddInt32(sc.currentConcurrency, 1) defer atomic.AddInt32(sc.currentConcurrency, -1) // 执行流式推理 stream : req.Model.InferStream(req.Prompt) for { token, err : stream.Next() if err io.EOF { break } sc.responseQueue - StreamResponse{ RequestID: req.ID, Token: token, Done: false, } } // 发送结束标记 sc.responseQueue - StreamResponse{ RequestID: req.ID, Done: true, } }2.3 并发控制策略func (sc *StreamController) Submit(req *InferenceRequest) error { select { case sc.requestQueue - req: return nil default: return fmt.Errorf(请求队列已满) } } func (sc *StreamController) GetConcurrency() int { return int(atomic.LoadInt32(sc.currentConcurrency)) } func (sc *StreamController) AdjustConcurrency(newMax int) { sc.maxConcurrency newMax // 动态调整工作协程数量 }三、 性能对比指标传统方案优化方案提升峰值 QPS3001200↑ 300%平均延迟500ms180ms↓ 64%P99 延迟1200ms350ms↓ 70.8%显存利用率波动大稳定在 85%-服务可用性95%99.99%↑ 5.3%四、 进阶优化请求优先级4.1 优先级队列实现type PriorityRequest struct { req *InferenceRequest priority int index int } type PriorityQueue []*PriorityRequest func (pq PriorityQueue) Len() int { return len(pq) } func (pq PriorityQueue) Less(i, j int) bool { return pq[i].priority pq[j].priority } func (pq PriorityQueue) Swap(i, j int) { pq[i], pq[j] pq[j], pq[i] pq[i].index i pq[j].index j } func (pq *PriorityQueue) Push(x interface{}) { n : len(*pq) item : x.(*PriorityRequest) item.index n *pq append(*pq, item) heap.Fix(pq, n) } func (pq *PriorityQueue) Pop() interface{} { old : *pq n : len(old) item : old[n-1] item.index -1 *pq old[0 : n-1] return item }4.2 请求调度策略graph LR A[高优先级请求] -- B[优先队列] C[普通请求] -- D[普通队列] B -- E[调度器] D -- E E -- F[工作协程]五、 实战技巧监控与调优5.1 关键指标监控from prometheus_client import Gauge, Counter # 并发数 concurrency_gauge Gauge(stream_concurrency, 当前并发数) # 请求计数 request_counter Counter(stream_requests, 请求总数, [status]) # 延迟直方图 latency_histogram Histogram(stream_latency, 响应延迟)5.2 调优参数参数默认值优化值说明max_concurrency1050根据 GPU 显存调整queue_size100010000缓冲突发请求priority_enabledfalsetrue开启优先级调度timeout30s60s流式请求超时时间六、 避坑指南6.1 响应乱序问题// 问题异步处理导致响应乱序 // 解决方案使用请求ID排序 type ResponseBuffer struct { mu sync.Mutex buffers map[string][]string callbacks map[string]func([]string) } func (rb *ResponseBuffer) Add(id string, token string, done bool) { rb.mu.Lock() defer rb.mu.Unlock() rb.buffers[id] append(rb.buffers[id], token) if done { cb : rb.callbacks[id] delete(rb.callbacks, id) cb(rb.buffers[id]) delete(rb.buffers, id) } }6.2 GPU 资源竞争// 使用 semaphore 控制 GPU 访问 var gpuSem semaphore.NewWeighted(int64(maxGPUConcurrency)) func inferWithGPU(ctx context.Context, req *Request) error { if err : gpuSem.Acquire(ctx, 1); err ! nil { return err } defer gpuSem.Release(1) // 执行 GPU 推理 return model.Infer(req) }总结三个核心优化点并发控制使用限流器控制最大并发数队列缓冲请求队列平滑流量峰值优先级调度保证重要请求优先处理从 300 QPS 到 1200 QPS4 倍吞吐提升。流式服务的并发控制是个精细活需要在吞吐量和稳定性之间找到平衡点。
高吞吐场景下大模型推理服务的流式并发控制设计
高吞吐场景下大模型推理服务的流式并发控制设计前言最近在做一个高吞吐的大模型推理服务要求支持 1000 QPS 的流式输出。一开始采用简单的 goroutine 并发模式结果遇到了严重的问题并发量上去后GPU 显存瞬间打满流式响应出现乱序服务在高负载下不稳定经过几轮优化最终实现了稳定的高吞吐流式服务。这篇文章分享完整的设计方案。一、 问题分析1.1 传统方案的瓶颈graph TD A[并发请求] -- B[goroutine] B -- C[GPU推理] C -- D[流式响应] E[无限制并发] --|导致 | F[显存爆炸] B -.- E1.2 性能数据指标传统方案问题峰值并发无限制显存 OOM响应顺序乱序客户端难以处理资源利用率不稳定GPU 利用率波动大服务可用性95%高负载下崩溃二、 解决方案流式并发控制器2.1 架构设计graph TD A[客户端请求] -- B[请求队列] B -- C[并发控制器] C -- D[GPU推理池] D -- E[响应缓冲] E -- F[流式输出] G[配置中心] -- C H[监控系统] -- C2.2 核心实现type StreamController struct { maxConcurrency int currentConcurrency int32 requestQueue chan *InferenceRequest responseQueue chan *StreamResponse wg sync.WaitGroup ctx context.Context cancel context.CancelFunc } func NewStreamController(maxConcurrency int) *StreamController { ctx, cancel : context.WithCancel(context.Background()) return StreamController{ maxConcurrency: maxConcurrency, requestQueue: make(chan *InferenceRequest, 10000), responseQueue: make(chan *StreamResponse, 10000), ctx: ctx, cancel: cancel, } } func (sc *StreamController) Start() { // 启动响应处理器 go sc.processResponses() // 启动工作协程 for i : 0; i sc.maxConcurrency; i { sc.wg.Add(1) go sc.worker() } } func (sc *StreamController) worker() { defer sc.wg.Done() for { select { case req : -sc.requestQueue: sc.processRequest(req) case -sc.ctx.Done(): return } } } func (sc *StreamController) processRequest(req *InferenceRequest) { // 原子增加并发计数 atomic.AddInt32(sc.currentConcurrency, 1) defer atomic.AddInt32(sc.currentConcurrency, -1) // 执行流式推理 stream : req.Model.InferStream(req.Prompt) for { token, err : stream.Next() if err io.EOF { break } sc.responseQueue - StreamResponse{ RequestID: req.ID, Token: token, Done: false, } } // 发送结束标记 sc.responseQueue - StreamResponse{ RequestID: req.ID, Done: true, } }2.3 并发控制策略func (sc *StreamController) Submit(req *InferenceRequest) error { select { case sc.requestQueue - req: return nil default: return fmt.Errorf(请求队列已满) } } func (sc *StreamController) GetConcurrency() int { return int(atomic.LoadInt32(sc.currentConcurrency)) } func (sc *StreamController) AdjustConcurrency(newMax int) { sc.maxConcurrency newMax // 动态调整工作协程数量 }三、 性能对比指标传统方案优化方案提升峰值 QPS3001200↑ 300%平均延迟500ms180ms↓ 64%P99 延迟1200ms350ms↓ 70.8%显存利用率波动大稳定在 85%-服务可用性95%99.99%↑ 5.3%四、 进阶优化请求优先级4.1 优先级队列实现type PriorityRequest struct { req *InferenceRequest priority int index int } type PriorityQueue []*PriorityRequest func (pq PriorityQueue) Len() int { return len(pq) } func (pq PriorityQueue) Less(i, j int) bool { return pq[i].priority pq[j].priority } func (pq PriorityQueue) Swap(i, j int) { pq[i], pq[j] pq[j], pq[i] pq[i].index i pq[j].index j } func (pq *PriorityQueue) Push(x interface{}) { n : len(*pq) item : x.(*PriorityRequest) item.index n *pq append(*pq, item) heap.Fix(pq, n) } func (pq *PriorityQueue) Pop() interface{} { old : *pq n : len(old) item : old[n-1] item.index -1 *pq old[0 : n-1] return item }4.2 请求调度策略graph LR A[高优先级请求] -- B[优先队列] C[普通请求] -- D[普通队列] B -- E[调度器] D -- E E -- F[工作协程]五、 实战技巧监控与调优5.1 关键指标监控from prometheus_client import Gauge, Counter # 并发数 concurrency_gauge Gauge(stream_concurrency, 当前并发数) # 请求计数 request_counter Counter(stream_requests, 请求总数, [status]) # 延迟直方图 latency_histogram Histogram(stream_latency, 响应延迟)5.2 调优参数参数默认值优化值说明max_concurrency1050根据 GPU 显存调整queue_size100010000缓冲突发请求priority_enabledfalsetrue开启优先级调度timeout30s60s流式请求超时时间六、 避坑指南6.1 响应乱序问题// 问题异步处理导致响应乱序 // 解决方案使用请求ID排序 type ResponseBuffer struct { mu sync.Mutex buffers map[string][]string callbacks map[string]func([]string) } func (rb *ResponseBuffer) Add(id string, token string, done bool) { rb.mu.Lock() defer rb.mu.Unlock() rb.buffers[id] append(rb.buffers[id], token) if done { cb : rb.callbacks[id] delete(rb.callbacks, id) cb(rb.buffers[id]) delete(rb.buffers, id) } }6.2 GPU 资源竞争// 使用 semaphore 控制 GPU 访问 var gpuSem semaphore.NewWeighted(int64(maxGPUConcurrency)) func inferWithGPU(ctx context.Context, req *Request) error { if err : gpuSem.Acquire(ctx, 1); err ! nil { return err } defer gpuSem.Release(1) // 执行 GPU 推理 return model.Infer(req) }总结三个核心优化点并发控制使用限流器控制最大并发数队列缓冲请求队列平滑流量峰值优先级调度保证重要请求优先处理从 300 QPS 到 1200 QPS4 倍吞吐提升。流式服务的并发控制是个精细活需要在吞吐量和稳定性之间找到平衡点。
