读完阿里百炼网关团队那篇《限流比降 10 倍百炼网关如何用 RocketMQ LiteTopic 重构大模型限流》给了我很大启发——网关还能这么玩。前半段复盘百炼限流的核心思路固定窗口漏桶、LiteTopic 分布式漏桶矩阵、Suspend 控速后半段补上原文跳过的关键一环消费端调模型后流式响应怎么回到客户端。这部分是我自己推的不是百炼的实现。大模型时代限流变了什么以前的限流主要解决防刷和防雪崩。CPU、内存、带宽都能快速扩处理起来比较直接。大模型不一样。GPU 扩容周期长、单价高供给受硬件交付节奏限制。百炼是阿里云的大模型服务平台百万级用户同时调千问等几十种大模型。用户从千级涨到百万级后限流要解决的问题变成了有限的 GPU 池子怎么分给百万租户还要各自独立、按需弹性。限流误判一次用户侧就是超时重试甚至业务中断过载放行一次就是稀缺 GPU 算力的浪费还殃及其他租户。算法选择固定窗口 漏桶百炼面对的不是单一限流场景是三类叠在一起SLA 基础限流要稳定、可计量、可观测User Model 维度的细粒度管控客户内部多账号也要交叉约束超大客户的突发承接——额度调到很大突发真来了 GPU 顶不住硬限大面积 503放过去又过载算法上最终选了固定窗口 漏桶固定窗口而不是滑动窗口因为滑动窗口太严格边界处的突发都会精确算到固定窗口对短期波动更宽容。漏桶而不是令牌桶因为漏桶匀速放行契合 GPU 稳态友好、尖峰敏感的特点。令牌桶天然允许突发正好是 GPU 最怕的形态。漏桶必须搬出进程算法定了新问题来了超大客户的突发流量堆在网关进程内存里几十万请求直接把网关自己打挂。漏桶必须从进程内搬到进程外用一个足够大、足够隔离的外部存储承接缓冲。把接住请求和按节奏放行在物理上拆开——这是 LiteTopic 进入视野的起点。传统 Topic / Group 撑不住百炼最早用传统 RocketMQ Topic 做漏桶每个 User Model 单独建 Topic 和 Consumer Group再部署一组消费机器。但头部客户接进来后三个痛点全暴露了元数据重、生效慢。传统 Topic/Group 要预先创建生效几十秒新用户突发流量到了还没建好。机器利用率低。共享消费模型要求同一 Consumer Group 下订阅完全相同的 Topic 集合否则堆积甚至丢消息。要隔离客户就得每个客户单独建 Group、单独切机器哪怕今天一条消息没有那组机器也常驻。单 Topic 暴涨株连。硬把多客户塞进同一 Topic某个用户消息激增几乎所有消费线程被它占住其他用户全堵。最后只能退回一客户一组机器用机器规模换隔离。传统方案能服务几个头部突发客户但没法让任意中大型客户都自动接入漏桶。限流要从VIP 待遇变成平台基础能力底层得重做。LiteTopic 的三个差异LiteTopic 是 RocketMQ 5.x 的轻量级队列跟传统 Topic 三个关键差异正好对应上面三个痛点。轻量元数据单 Broker 撑百万级 LiteTopic运行时按需创建、按 TTL 自动回收。客户端往不存在的 LiteTopic 发消息就自动建出来没活动一段时间 Broker 自动清理。差异化订阅同一 Consumer Group 下不同实例可以订阅不同的 LiteTopic 子集不会堆积也不会丢。所有租户共享一组消费 Pod不再需要一客户一组机器。Suspend 消费控制消费回调里返回Suspend N毫秒Broker 就在 N 毫秒内暂停拉取这个 LiteTopic不影响其他 LiteTopic。这是漏桶在 Broker 层落地的关键开关。重构后的架构每个 User Model 一个独立漏桶桶容量由 LiteTopic 堆积上限和 TTL 决定出水速率由消费侧 Suspend 时长动态决定。所有漏桶共享一组消费机器资源随总流量伸缩不按客户数线性增长。Suspend 到底做了什么架构图里那行命中限流 → return Suspend(N),Broker 暂停该 LiteTopic展开讲一下。消费端拉到一条消息后先跑限流判断。判断结果是要么放行、要么暂停。如果暂停消费端不调模型、不 ACK而是告诉 Broker这个 LiteTopic 的下一条消息N 毫秒之后再给我。然后消费线程立刻返回去拉别的 LiteTopic 的消息。关键在于暂停是 Broker 侧的不是消费端自己 sleep。Broker 收到 Suspend(N) 后在这 N 毫秒内不会再把这个 LiteTopic 的消息投递给这个消费端。但同一消费端对其他 LiteTopic 的拉取完全不受影响——其他 LiteTopic 有消息照样拉、照样消费。时刻 0ms: 消费端拉到 liteTopic-A-X 的一条消息 限流判断:需要暂停 消费端 return Suspend(500) → 消费线程立刻释放 同时: 消费端拉到 liteTopic-B-Y 的一条消息 ← 不受影响 限流判断:放行 消费端调用模型,正常 ACK 时刻 500ms: Broker 恢复对 liteTopic-A-X 的投递 消费端再次拉到这条消息 限流判断:可以放行了 消费端调用模型,正常 ACK对比一下如果用Thread.sleep(500)会怎样消费线程卡在 sleep 上 500 毫秒不动这 500 毫秒内这个线程既不消费 liteTopic-A-X 的后续消息也不消费其他任何 LiteTopic 的消息。如果被限流的 LiteTopic 多了线程池被占满没被限流的租户也跟着堆积。原文有个 POC 对比订阅 50 个 LiteTopic前 5 个随机触发 300ms 限流。Thread.sleep(300)时被限流的各占一个线程不释放限流数增到 40 个时线程池几乎耗尽剩下 10 个正常的全堆积。Suspend(300)时消费线程返回后立即释放回线程池转头服务其他 LiteTopic被 Suspend 的由 Broker 在 300ms 后重投结果只有被限流的 5 个堆积其余 45 个正常。所以 Suspend 和 Sleep 的区别不是等不等——两个都要等 300ms。区别是等待期间那个消费线程能不能去干别的。Suspend 把等这件事交给 Broker 计时消费线程不等立刻去服务其他租户。这个机制也是漏桶匀速放行的物理实现。假设某个 User Model 的配额是每秒 10 个请求消费端每次放行一条消息后return Suspend(100)Broker 每 100ms 才投递下一条正好每秒 10 条。配额变了改 Suspend 的 N 就行不用重启、不用改配置。当前 Suspend 最小粒度是 30 毫秒。更细的速率控制需要消费端自己局部 sleep 补充。百炼的实际做法是 Suspend 承担主要漏桶调速极少数亚 30ms 精度场景用 Sleep 补充。为什么 LiteTopic 适合当漏桶桶的容量工程意义上近似无限。进程内队列容量是写死的数字桶满即拒。LiteTopic 持久化到 Broker 磁盘单实例百万级队列堆积上限远超进程内队列彼此物理隔离。客户尖峰被接住、攒起来、按节奏放行等几秒再处理几乎总比直接 429 好。每个 User Model 一个独立漏桶。隔离变成了命名规则问题客户 A 的 Model X 进liteTopic-A-X客户 B 的进liteTopic-B-XBroker 层物理隔离A 堵了不会传染到 B。租户隔离从靠业务代码维护变成靠基础设施天然提供。每个漏桶速率可独立动态调整。Suspend N由业务策略实时算毫秒级调整任意一个漏桶的放行速度重要客户在模型负载高时仍能拿到较高节奏模型扩容到位后矩阵速率同步上调某款模型有拥塞苗头单独收紧那一列漏桶就行。不重启消费组、不改配置、不需要客户感知。原文没展开的问题响应怎么回去文章把核心改动浓缩成三段代码消费回调这段最关键OverridepublicConsumeResultconsume(MessageViewmsg){StringliteTopicmsg.getLiteTopic();longsuspendMsrateLimitPolicy.acquireOrSuspend(liteTopic);if(suspendMs0){returnConsumeResult.Suspend(suspendMs);}invokeModel(msg);// 不限流就直接调模型returnConsumeResult.SUCCESS;}invokeModel(msg)这一行原文没展开。下面的分析是我自己推的不是百炼的实现——原文没有给出响应回传的细节这里是我根据这个架构的约束推导出来的一个可行方案。问题是这样的客户端的 HTTP 连接握在生产端网关入口模型调用却发生在消费端另一组 Pod。LLM 响应又是流式的一个个 token 往外吐。消费端拿到的流式响应怎么实时回到生产端那条还挂着的客户端连接上还有个硬约束消费线程很宝贵POC 才 50 个不能让一个流式响应占住线程 20 秒什么都不干。为什么响应通道不用 LiteTopic既然请求通道已经用了 LiteTopic响应通道为什么不再用一个因为两个通道要的东西不一样。请求通道要限流Suspend 控速、要缓冲堆积削峰、要隔离百万租户、可以等几十秒。响应通道这些一个都不需要——chunk 来了就该立刻送读完即删天然一对一按 reqId每个 chunk 多几毫秒延迟用户都能感知到。LiteTopic 的三件套能力是给请求通道量身定做的响应通道用不上。LLM 逐 token 吐每个 token 间隔几十毫秒如果响应通道每个 chunk 都走一遍 Broker 的完整链路路由、持久化、就绪集合通知每个 token 多 3-5ms。一条响应几百个 token累积下来流畅度明显下降。Redis 的 XADD/XREAD 是内存操作亚毫秒级没有多余动作。还有队列数量的问题。百炼已经百万级 LiteTopic响应通道再建一个每个请求一个Broker 承载翻倍到 200 万队列。Redis 这边百万个 key 就是一堆字符串内存占用小一个数量级。用 Redis Streams 做响应通道我的思路是用 Redis Streams 做生产端和消费端之间的响应管道每个请求一个独立 Streamkey 是response:{reqId}。整条链路有两条通道请求走 LiteTopic响应走 RedisreqId 把它们串起来。流程按编号看客户端发 HTTP 请求到生产端生产端生成 reqId生产端把请求消息连同 reqId 写入 LiteTopic按 UserModel 路由消费端从 LiteTopic 拉到消息命中限流则return Suspend(N)等待不命中继续消费端用 reqId 作为关联标识向 GPU 推理服务发起流式调用GPU 逐 token 返回响应消费端每收到一个 chunkXADD 写入response:{reqId}这个 Redis Stream生产端一直在用 XREAD 阻塞读这个 Stream收到 chunk 立刻推给客户端生产端通过 SSE 把 chunk 流式推回客户端循环直到收到 done 标记两条通道各管各的LiteTopic 管请求按什么节奏放行给 GPURedis Stream 管响应以什么延迟回到客户端。reqId 贯穿全链路生产端生成 reqId随消息写进 LiteTopic 的 header。消费端从消息里取出来用同样的规则拼出 Redis Stream key 去写响应。生产端用同一个 key 去读。两边不需要额外通知key 规则是约定好的reqId 对上就行。生产端核心XREAD 阻塞循环生产端写完 LiteTopic 后从 Redis Stream 阻塞读响应。XREAD 的 BLOCK 参数让等待过程不耗 CPU数据来了自动唤醒。还加了一个总超时防止消费端崩溃后生产端无限等func(g*GatewayProducer)HandleRequest(w http.ResponseWriter,r*http.Request){reqID:uuid.New().String()streamKey:response:reqID// 构造消息,reqId 塞进 header,按 usermodel 路由到 LiteTopicmsg:primitive.Message{Topic:g.parentTopic,Body:payload}msg.SetLiteTopic(fmt.Sprintf(litetopic-%s-%s,user,model))msg.WithProperty(reqId,reqID)g.mqProducer.SendSync(r.Context(),msg)// 开启 SSE,从 Redis Stream 流式读响应推给客户端w.Header().Set(Content-Type,text/event-stream)flusher,_:w.(http.Flusher)lastID:0// 从头读,不漏第一条 chunk// 总超时:消费端崩溃或 Redis 挂了时,防止生产端无限等// LLM 流式响应一般不超过 60 秒,120 秒兜底ctx,cancel:context.WithTimeout(r.Context(),120*time.Second)defercancel()for{// XREAD BLOCK:没数据卡住等,有数据立刻返回result,err:g.rdb.XRead(ctx,redis.XReadArgs{Streams:[]string{streamKey,lastID},Count:10,Block:30*time.Second,}).Result()iferrredis.Nil{continue// 超时无数据,继续等(消息可能还在 Suspend 排队)}iferr!nil{// context deadline 超时,或客户端断连,或 Redis 连接异常ifctx.Err()context.DeadlineExceeded{fmt.Fprintf(w,event: timeout\ndata: response incomplete\n\n)flusher.Flush()}return}for_,stream:rangeresult{for_,message:rangestream.Messages{lastIDmessage.ID// 推进游标,下次从这里之后读switchmessage.Values[type]{casechunk:fmt.Fprintf(w,data: %s\n\n,message.Values[chunk])flusher.Flush()// 立即推给客户端casedone:fmt.Fprintf(w,event: done\ndata: [DONE]\n\n)flusher.Flush()returncaseerror:fmt.Fprintf(w,event: error\ndata: %s\n\n,message.Values[error])return}}}}}XREAD 按 ID 游标拉取告诉 Redis 上次读到哪个 ID它返回那个 ID 之后的所有新消息没有就阻塞等。lastID 是生产端自己存在内存里的每读到一条就往前推不重复、不遗漏、顺序一致。Count: 10限制单次最多返回 10 条防止消费端瞬间写入大量 chunk 时一次性全部返回导致处理卡顿。这个值不影响正确性——lastID 游标保证不丢消息设 5 或 20 都行。这个 for 循环不会死循环也不会空转占 CPU。不死循环是因为有四个出口收到 done 正常 return、收到 error 异常 return、客户端断连时r.Context()自动 cancel 导致 XREAD 返回错误走if err ! nil分支 return、总超时 120 秒到了 context deadline exceeded 也 return。不占 CPU 是因为XREAD BLOCK 30000是服务端阻塞goroutine 挂起在 IO 等待上不消耗 CPURedis 服务端没数据时也把连接挂起等新消息唤醒。30 秒没数据返回 Nilcontinue 再发一次 XREAD又挂起等。百万并发就是百万个 goroutine 各自挂起等自己的 keyCPU 消耗接近零。总超时是兜底措施。消费端崩溃、Redis 挂了、消息进了死信队列——这些情况最终都表现为生产端收不到 done。没有总超时生产端会一直 continue 下去客户端无限挂起。120 秒内没收到 done就认定消费端出了问题给客户端返回 timeout 事件。需要注意的是如果消费端 Suspend 很久比如限流 60 秒期间 HTTP 连接一直挂着网关前面的 LB、Nginx 超时设置要大于最大 Suspend 时长否则 LB 先把连接断了。消费端核心回调里调模型chunk 写 Redis消费端除了限流判断和幂等检查还加了processing:{reqId}标记防止消息重投时重复调模型以及 WriteChunk 失败时中断模型调用func(c*GatewayConsumer)Consume(ctx context.Context,msgs...*primitive.MessageExt)(consumer.ConsumeResult,error){for_,msg:rangemsgs{reqID:msg.GetProperty(reqId)liteTopic:msg.GetLiteTopic()// 限流判断:返回 0 表示需要 SuspendsuspendMs:c.rateLimitPolicy(liteTopic)ifsuspendMs0{returnconsumer.Suspend(suspendMs),nil// 消费线程立刻释放}// 幂等检查:消息重投时防止重复调模型// done 标记:上次已经完整处理过ifexists,_:c.rdb.Exists(ctx,done:reqID).Result();exists0{continue}// processing 标记:上一次调了一半崩了,还没写 done// 重投后不应该重新调模型,否则客户端会收到两段拼起来的响应ifexists,_:c.rdb.Exists(ctx,processing:reqID).Result();exists0{// 写 error 通知客户端重新发起请求WriteError(ctx,c.rdb,reqID,previous attempt incomplete, please retry)continue}// 标记正在处理,120 秒后自动过期(兜底,防止崩了不清理)c.rdb.Set(ctx,processing:reqID,1,120*time.Second)// 流式调用模型,每个 chunk 写 Redis Streamerr:c.modelClient.Invoke(ctx,model,prompt,func(chunkstring)error{// 检查客户端是否已断连ifn,_:c.rdb.Exists(ctx,cancel:reqID).Result();n0{returnfmt.Errorf(client disconnected)// 中断模型调用,省 GPU}// WriteChunk 失败说明 Redis 挂了,响应回不去,继续调模型也是白花 GPUiferr:WriteChunk(ctx,c.rdb,reqID,chunk);err!nil{returnfmt.Errorf(redis unavailable, abort model call: %w,err)}returnnil})iferr!nil{// 模型调用被中断(客户端断连或 Redis 挂)// 写 error 通知生产端(如果 Redis 还活着的话),然后 ACK 当前消息// 当前消息处理完毕(含失败),继续处理下一条WriteError(ctx,c.rdb,reqID,err.Error())c.rdb.Del(ctx,processing:reqID)continue}// 写 done 标记,清理 processing 标记,ACKWriteDone(ctx,c.rdb,reqID)c.rdb.Del(ctx,processing:reqID)}returnconsumer.ConsumeSuccess,nil}消费端多了两个防御。processing:{reqId}标记解决的是调模型中途崩溃消息重投后重新调模型的问题——上一次已经往 Redis Stream 写了几个 chunk客户端收到了部分响应重投后再调一次模型客户端会收到两段拼起来的响应。加上这个标记后重投时发现有 processing 标记说明上次没干完写一个 error 让客户端重新发起请求而不是重复调用。WriteChunk 失败中断模型调用解决的是Redis 挂了但还在白花 GPU的问题。Redis 挂了之后消费端写的 chunk 没人能读到继续调模型纯粹浪费算力。回调返回 errorInvoke 中断不 ACKRedis 挂了算基础设施异常让 MQ 重投等 Redis 恢复。变体生产端和消费端同进程时用 channel 替代 Redis前面假设生产端和消费端是分开的 Pod响应必须跨进程传递所以用了 Redis Streams。但如果生产端和消费端在同一个进程里——网关进程既负责接收 HTTP 请求写 LiteTopic又作为消费端从 LiteTopic 拉消息调模型——响应通道根本不需要 Redis用 Go channel 在进程内传递就行。请求还是走 LiteTopic限流、缓冲、隔离这些能力仍然需要 Broker 提供只是响应从跨进程走 Redis变成进程内走 channel同一个网关进程 ┌──────────────────────────────────────────────────────┐ │ │ │ HTTP Handler (goroutine A) Consumer (goroutine B)│ │ ┌──────────────┐ ┌──────────────────┐ │ │ │ 生成 reqId │ │ 拉到消息 │ │ │ │ 写 LiteTopic │ │ 限流判断 │ │ │ │ │ │ 调模型 │ │ │ │ 等 chan │ ◄─────────── │ chunk 写 chan │ │ │ │ 推给客户端 │ Go channel │ done 写 chan │ │ │ └──────────────┘ └──────────────────┘ │ │ │ │ ┌─────────────────────┐ │ │ │ sync.Map │ │ │ │ reqId → chan chunk │ │ │ └─────────────────────┘ │ └──────────────────────────────────────────────────────┘生产端用 reqId 注册一个 channel消费端通过 reqId 找到这个 channel 直接写// 进程内响应注册表varresponseChans sync.Map// reqId - chan StreamEventtypeStreamEventstruct{Typestring// chunk / done / errorChunkstringErrorstring}// 生产端:注册 channel,写 LiteTopic,然后阻塞等func(g*Gateway)HandleRequest(w http.ResponseWriter,r*http.Request){reqID:uuid.New().String()ch:make(chanStreamEvent,100)// 带缓冲,消费端写不阻塞responseChans.Store(reqID,ch)deferresponseChans.Delete(reqID)// 写 LiteTopic(跟之前一样)msg:primitive.Message{Topic:g.parentTopic,Body:payload}msg.SetLiteTopic(fmt.Sprintf(litetopic-%s-%s,user,model))msg.WithProperty(reqId,reqID)g.mqProducer.SendSync(r.Context(),msg)// SSE,从 channel 读响应推给客户端w.Header().Set(Content-Type,text/event-stream)flusher,_:w.(http.Flusher)ctx,cancel:context.WithTimeout(r.Context(),120*time.Second)defercancel()for{select{case-ctx.Done():fmt.Fprintf(w,event: timeout\ndata: response incomplete\n\n)flusher.Flush()returncaseevent:-ch:switchevent.Type{casechunk:fmt.Fprintf(w,data: %s\n\n,event.Chunk)flusher.Flush()casedone:fmt.Fprintf(w,event: done\ndata: [DONE]\n\n)flusher.Flush()returncaseerror:fmt.Fprintf(w,event: error\ndata: %s\n\n,event.Error)return}}}}// 消费端:拉到消息后,通过 reqId 找到 channel 直接写func(g*Gateway)Consume(ctx context.Context,msgs...*primitive.MessageExt)(consumer.ConsumeResult,error){for_,msg:rangemsgs{reqID:msg.GetProperty(reqId)liteTopic:msg.GetLiteTopic()suspendMs:g.rateLimitPolicy(liteTopic)ifsuspendMs0{returnconsumer.Suspend(suspendMs),nil}// 从注册表找到对应的 channelval,ok:responseChans.Load(reqID)if!ok{continue// 生产端已经走了(客户端断连),跳过}ch:val.(chanStreamEvent)// 调模型,每个 chunk 直接写 channelerr:g.modelClient.Invoke(ctx,model,prompt,func(chunkstring)error{// 检查 channel 是否还在(客户端断连后生产端会 defer Delete)_,ok:responseChans.Load(reqID)if!ok{returnfmt.Errorf(client disconnected)}ch-StreamEvent{Type:chunk,Chunk:chunk}returnnil})iferr!nil{ch-StreamEvent{Type:error,Error:err.Error()}}else{ch-StreamEvent{Type:done}}}returnconsumer.ConsumeSuccess,nil}channel 方案不需要 Redis不需要 XADD/XREAD不需要 Stream key 的 TTL 清理不需要 done/processing/cancel 这些 Redis 标记。客户端断连就是生产端 goroutine 退出defer responseChans.Delete(reqID)自动清理消费端Load找不到 channel 就知道客户端走了直接中断模型调用。少了一整个 Redis 依赖。但代价也明显。生产端和消费端绑在同一个进程里进程挂了两边一起没。Redis 方案里消费端挂了MQ 重投给另一个消费实例就行同进程方案下进程重启后在途的请求全丢了客户端只能靠总超时感知到然后重试。另外消费端调模型如果涉及 CPU 密集的预处理比如图片编码可能跟 HTTP handler 抢 CPU分进程方案没这个问题。中小规模——租户几千到几万——同进程 channel 够用少一个 Redis 依赖。百炼这种百万租户规模生产端和消费端必须分开部署独立伸缩Redis Streams 是更现实的选择。几个工程细节ACK 时机invokeModel之后紧接着 ACK但要分情况模型调用成功 → 写 done → ACK 模型业务失败 → 写 error → ACK(不重投,重试只会再花一次 GPU 算力) Redis/基础设施异常 → 不 ACK → Broker 重投模型业务错误不重投重试结果大概率一样白花 GPU。只有基础设施异常才重投。幂等性Broker 和消费端之间 ACK 可能丢同一条消息被投两次。消费端代码里有两层检查done:{reqId}标记防的是已完整处理的请求被重投直接跳过不重复调模型processing:{reqId}标记防的是调了一半崩了的请求被重投写 error 让客户端重新发起不重复调模型。LLM 推理一次消耗数秒 GPU 时间重复调用是纯浪费。客户端断连客户端关了浏览器生产端检测到连接断开Go 里r.Context().Done()自动触发往cancel:{reqID}写取消标记。消费端在每个 chunk 之间检查发现取消就中断模型调用。不做这个处理用户断连后模型还在推理一次调用几秒到几十秒 GPU 时间就白花了。小结百炼从进程内漏桶走到分布式漏桶矩阵核心不是算法多新颖——固定窗口和漏桶都是几十年的老东西。真正的问题是规模到了百万租户之后瓶颈不在算法在于你用什么物理载体去承载那个拆出来的结构。算法选型之外漏桶的物理载体、租户隔离的成本模型、响应通道的延迟特性这些才是决定方案能不能 scale 的东西。文章把响应回传这块全部藏在了invokeModel(msg)一行里但这是整个方案能不能真正跑起来的关键。响应通道的选择取决于部署形态生产端和消费端同进程时用 Go channel进程内传递最简单分进程部署时用 Redis Streams跨进程传递但有容错。请求走 LiteTopic 做限流缓冲响应按部署形态选 channel 或 Redis两条通道各干各的不要强行统一。Suspend vs Sleep 的选择、消息重投的幂等处理、生产端总超时兜底、客户端断连时中断模型调用——这些看起来是细节但在百万租户场景下任何一个出问题都是生产事故。百炼原文聚焦在限流架构本身这些工程细节需要自己补上。参考资料限流比降 10 倍百炼网关如何用 RocketMQ LiteTopic 重构大模型限流 — 阿里云千问AI平台PDF 存档RocketMQ 5.x LiteTopic 文档 — Apache RocketMQRedis Streams 官方文档 — Redisgo-redis 文档 — go-redis构建高效业务网关基于 Gin 与 ReverseProxy 的实现方案 — 本博客从模型管理到 AI 网关把资产、流量与成本串成闭环 — 本博客总量节流之外的另一条限流线按令牌分账的裁决