Android 流式对话实现：分层门面与 SSE 手工解析

上层只管刷新哪条气泡下层只管逐行读字节——把传输细节彻底挡在业务之外前言做流式问答时最常踩的坑不是发请求而是流式回答来了不知道该刷哪条气泡。用户已经发出消息界面立刻生成一个正在输入的占位气泡大模型一个字一个字往回吐吐到一半停住了我们要么整个界面卡死要么消息错位、张冠李戴。另一头的现象更反常识接口明明是流式的界面却是转半天圈后整段哐一下冒出来。这其实是两处不同层面的问题。往上看是业务调度层缺少流分片该写进哪个气泡的映射往下看是网络层的字节流被客户端的读取超时或日志拦截器截胡了。很多人把它们混在一起排查越查越乱——一会儿怀疑服务器没流式返回一会儿又去改界面刷新逻辑其实都没打在点上。这篇我把这两层拆开讲上面用分层门面把传输细节彻底挡在业务之外下面再把门面挡住的那层——SSE 传输——拎出来讲透。你看完会明白流式聊天最难的不是流本身而是流的每一段该写进哪个气泡上以及怎么保证那段流别在客户端被偷偷掐掉。一、现象流式对话最常见的四种翻车流式接口的翻车现场大致分两类“碎片找不到家和流退化了”。1.1 碎片找不到家多气泡互相串台传统一问一答的接口是同步的请求发出去等一个完整的 JSON 回来一次性渲染。可大模型聊天是流式的服务器会像打字机一样分多次把同一段回答的碎片推过来。于是出现一个尴尬的局面界面上有好几个正在回复的气泡用户可能连发两条、或者打断重问服务器推回来的碎片到底属于哪一条用户帮我查一下设备状态 → 界面生成气泡 Aloading 用户算了换个角度再问一次 → 界面生成气泡 Bloading 服务器流式碎片1属于 A 服务器流式碎片2属于 B如果只靠按顺序对应两次请求几乎同时发出、返回交错时气泡 A 和 B 的内容就会互相串台。这是流式对话最反直觉、也最容易做错的地方。1.2 流退化了要么被缓存成整段要么被读到一半掐断另一类现象更隐蔽功能看起来能用只是没有流式效果或者偶尔残缺。典型表现是现象A整段一次性冒出无流式→ 客户端把流缓存了 现象B读到一半断流 → 客户端把流超时了代码明明用的是流式接口可表现却是转圈转半天然后哐一下整段回答同时冒出来——没有打字机效果。或者更糟读到一半连接被中断回答残缺。排查方向往往被带偏到是不是服务器没流式返回。但多数时候问题出在客户端两个现象一个指向日志拦截器一个指向读取超时都不是协议问题。二、根因两处信息断点把两类现象分开看根因就清楚多了调度层少了一张表网络层多了一次不该有的读取。2.1 调度层缺少流分片 ↔ 界面气泡的映射问题不在网络层而在调度层缺少一张映射表。服务器其实给每个会话/每个回复都打了唯一标识taskId而界面上每个气泡也有自己的消息 idmsgId。两头都有身份证但没人把这两个 id 对应起来碎片到了就不知道往哪塞。服务器侧taskId t1 → 这一段回复属于哪个会话 界面侧 msgId m1 → 这一条气泡是哪个 缺失t1 ↔ m1 的对应关系另一个附带问题回复是流式增量的界面不能等全部收完再显示必须把累积的文本边收边刷新。这意味着状态必须是可累积的而不是一锤子买卖。2.2 网络层字节流被截胡或掐断2.2.1 为什么日志拦截器会毁掉流式在 HTTP 客户端里加一个打印请求体/响应体的日志拦截器很常见。但流式响应下日志拦截器要打印完整响应体就得先把整个响应读进内存。这一读就把流给吞了——等它打印完原本逐帧到达的流已经全部读完后面真正消费流的地方拿到的是一整块内存数据流式效果荡然无存。正常 逐行读 → 逐行吐打字机 加日志整段读进内存 → 打印 → 一次性吐退化成非流式2.2.2 为什么读取超时等于定时炸弹普通接口设个 10 秒读取超时很合理。但流式对话是长时间挂着的连接两段事件之间可能间隔很久大模型在思考、在检索知识库。一旦设置了读取超时等待下一帧时超时器到期连接被强制关闭——回答就被掐在中间。这两处根因正对应标题里的两件事门面把映射理清楚SSE 解析把字节流稳定读出来。下面先讲门面。三、上半场分层门面把传输细节挡在业务之外3.1 四层门面各管一段把整个对话能力拆成四层越往上越接近用户越往下越接近网络界面层只展示气泡列表监听事件流不做任何网络逻辑调度层视图模型编排业务维护 taskId↔msgId 映射决定这个事件该刷新哪个气泡数据层仓库负责发起请求、解析流把原始字节转成结构化事件网络层封装 HTTP 客户端与接口定义。界面层永远只看到两类东西不可变的事件流和方法调用。它不知道也不关心大模型是谁、走什么协议。3.2 不可变事件流让界面只订阅、不操办界面需要的不是命令而是通知。所以调度层对外只暴露只读的asSharedFlow()内部才持有可写的MutableSharedFlow// 对外只读界面只能订阅valchatUIEvent_chatUIEvent.asSharedFlow()// 内部可写业务逻辑往里推事件privateval_chatUIEventMutableSharedFlowChatUiEvent(extraBufferCapacity1,// 事件可能来不及消费留一点缓冲onBufferOverflowBufferOverflow.DROP_OLDEST// 聊天气泡刷新允许丢最旧的)事件本身用密封结构描述开始流式回复 / 流式过程中 / 流式结束 / 出错各是一种事件携带这次回复属于哪个气泡的 id。界面只需when分派。3.3 那张关键映射表taskId ↔ msgId这是整个方案的精髓。用户一发消息界面立刻生成占位气泡并拿到msgId随后请求发出时把这个msgId作为自定义输入一起带给服务器。服务器每次流式推送都会回带taskId调度层在工作流开始事件里抓到这个taskId把taskId ↔ msgId写进一张内存表// 收到工作流开始建立映射difyTaskIdToMsgIdMap[taskId]respMsgId// 后续每个流式碎片/结束/错误事件都靠 taskId 反查该刷哪个气泡when(response.event){Message-{answerBuilder.append(response.answer)// 累积文本_chatUIEvent.tryEmit(Streaming(textanswerBuilder.toString(),// 推累积后的整段界面直接覆盖responseMsgIddifyTaskIdToMsgIdMap[response.taskId]!!))}MessageEnd-{// 流式结束附带引用文档等信息_chatUIEvent.tryEmit(StreamingEnd(refDocs...,responseMsgIddifyTaskIdToMsgIdMap[response.taskId]!!))}Error-{_chatUIEvent.tryEmit(Error(...,difyTaskIdToMsgIdMap[response.taskId]?:))}}注意一个细节推给界面的是累积后的完整文本而不是增量片段。界面拿到的永远是到目前为止的完整回答直接覆盖气泡即可。这避免了界面自己拼字符串、拼错还要纠正的麻烦。3.4 停止响应也能精准定位打断停止生成同样依赖映射表。用户点停止界面传msgId调度层反查taskId再调停止接口// 用 msgId 反查 taskId因为停止接口是按 taskId 调的difyTaskIdToMsgIdMap.forEach{(taskId,msgId)-if(msgIdresponseMessageId){stopChatMessage(taskId,...)}}四、下半场被门面挡住的 SSE长什么样门面把传输细节挡在了门外但挡在门外不等于不用懂真出问题时还是得掀开门帘看里面。SSEServer-Sent Events是 HTML 标准里定义的服务器推送机制MIME 类型是text/event-stream服务器把事件一行一行发过来客户端逐行读、逐行解析。真正让人翻车的往往不是协议本身而是 2.2 里那两个隐藏的坑。4.1 SSE 分帧规则不是读到一行就是一条事件按 HTML 标准Server-sent events 一节的规定分帧规则可以拆成这几条整体按 UTF-8 解码流开头的单个 BOMUFEFF会被忽略按行解析行结束符可以是CRLF、单独的LF或单独的CR以冒号:开头的行是注释直接忽略——服务器常用它发心跳保活例如: keep-alive字段行形如field: value冒号后如果紧跟一个空格这个空格会被去掉如果一行里没有冒号则整行当作字段名、值为空空行 事件边界遇到空行或流结束时派发事件。若 data 缓冲为空则不派发否则去掉末尾换行后作为一条消息。标准字段有四个data:数据、event:事件类型缺省为message、id:最后事件 ID、retry:重连间隔毫秒数。要注意第 5 条的反面一条事件可由多个data:行组成规范要求用\n拼接后再处理只是多数大模型服务每个事件只发一行data才让逐行取 data成了够用的简化做法。4.2 客户端能拿到的原始字节源是什么在 OkHttp 里Response.body是ResponseBody它的source()返回一个BufferedSourceOkio 的缓冲字节源也就是我们真正要逐行读的东西。两个必知事实ResponseBody是一次性的source()、string()、bytes()都会消费它string()/bytes()还会把整个响应读进内存并关闭它——这是退化成非流式的另一条路径流式场景必须走source()。readUtf8Line()读到换行或 EOFEOF 时返回null是阻塞调用会一直等到有数据或流结束——这正是它必须放在 IO 线程、且要能用call.cancel()打断的原因。五、合体逐行解析 SSE 的完整实现5.1 客户端配置valclientOkHttpClient.Builder().readTimeout(0,TimeUnit.SECONDS)// 流式必须读取永不超时.connectTimeout(30,TimeUnit.SECONDS)// 建连超时仍要.addInterceptor{chain-// 业务头注入在这里做但绝不打印响应体valbuilderchain.request().newBuilder()builder.addHeader(Authorization,Bearer$apiKey)chain.proceed(builder.build())}// 千万别加 BODY 级别的日志拦截器会把流缓存成整段.build()readTimeout(0)关闭读取超时流式连接想挂多久挂多久日志拦截器只打请求、不碰响应体如果要调试在消费流的地方逐条打印解析结果而不是拦截器里打整段建连超时仍然要设connectTimeout只在建连阶段生效与流式长连接不冲突。5.2 逐行解析核心就一个 while 循环回到分帧规则我们关心的只是data:行取出来就是 JSON。funchatStream(request:Request):FlowChunkflow{valresponseclient.newCall(request).execute()if(!response.isSuccessful){throwHttpException(response)}valbodyresponse.body?:throwIllegalStateException(Empty body)// 关键拿到原始字节源逐行读边读边 emitbody.source().use{source-while(!source.exhausted()){vallinesource.readUtf8Line()?:continueif(!line.startsWith(data:))continue// 只关心 data 行valjsonline.removePrefix(data:).trim()if(json[DONE])break// 结束标记if(jsonevent: ping)continue// 心跳事件跳过emit(gson.fromJson(json,Chunk::class.java))}}}.flowOn(Dispatchers.IO)几个要点readUtf8Line()是阻塞的所以整体必须跑在 IO 线程flowOn(Dispatchers.IO)否则会卡主线程只看data:前缀event:/id:行直接忽略——我们的业务只关心数据[DONE]是结束哨兵遇到就跳出循环心跳事件单独跳过避免误当成业务数据去反序列化报错每个data:帧就是一个独立 JSONemit出去就是流式的一次推进。要诚实说明的是[DONE]、event: ping这类标记不是 SSE 规范的一部分而是各家服务端自己的约定写法并不统一接入前应以服务端文档为准核对结束标记与心跳形式。5.3 线程分离流在 IO 线程读界面在哪刷新解析循环在 IO 线程逐个emit。这些流式推进必须最终回到主线程刷新 UI。用协程流的话就是在消费端用withContext(Dispatchers.Main)切回主线程后再更新界面或者让调度层在订阅时把线程切好。核心是读流在 IO渲染在主线程二者不能混。5.4 一个容易被漏掉的边界取消不会自动唤醒阻塞读emit挂起时会响应协程取消但若协程正卡在readUtf8Line()上阻塞读并不知道自己被取消了。要做到点停止就立刻断开可靠做法是持有这次调用的Call取消时主动call.cancel()关掉底层连接让阻塞读抛错返回——只依赖协程取消不够。TCP 粘包/拆包则交给BufferedSource处理应用层不必自己切行。一个可选的工程化增强把按行切分抽成可复用的 SSE 解析器输入字节流、输出事件行序列上层只管when分派。六、升华这是一种边界设计的通用思路这套方案不止于跑通一个聊天框它把两类边界都画清楚了门面把业务与传输的边界画在数据层SSE 解析把协议与应用的边界画在解析器。先说门面那半边。双标识 映射表的思路不限于大模型聊天任何异步多实例 回执需要归位的场景都适用多设备并发指令同一界面同时给多台设备发指令回执靠设备 id 归位批量任务进度多个上传/下载任务并行进度回调靠任务 id 找对应条目多通道推送同一消息从不同通道回来靠通道标识去重、聚合。核心要点就一个上游给每个实例打唯一标识下游用标识反查归属而不是依赖先来后到的顺序假设。顺序会交错id 不会。另外对外只读流、对内可写流的封装让界面代码极干净——它只订阅、只渲染永不直接触碰网络。想换一个问答服务商只动数据层和网络层界面一行不改。再说解析那半边。很多人把流式问答和大模型强绑定其实 SSE 就是个通用的服务器推送通道跟具体内容无关LLM 流式补全大模型逐字生成实时日志推送服务端把日志一行行推给前端知识库检索过程推送把开始检索 → 检索到文档 → 生成回答每个阶段推出来。只要掌握了逐行读 → 按前缀分拣 → 遇到哨兵收尾这个套路任何 SSE 服务都能吃下来。而且手工解析比引库更可控你能精确处理心跳、哨兵、脏数据还能按业务定制。结论流式对话的真正难点是碎片归位不是流本身用一张taskId ↔ msgId映射表解决该刷哪条气泡。用 MVVM 分四层界面只消费不可变事件流不碰网络天然解耦、可替换读流放 IO 线程、渲染回主线程两者不混。推给界面的用累积后的完整文本界面只覆盖不拼接避免自维护增量状态出错。打断/停止操作同样走映射表反查保证精确定位到目标实例。这类上游打 id、下游按 id 归位的思路可迁移到多设备、批量任务、多通道等异步场景。SSE 就是逐行文本协议一个readUtf8Line()循环即可解析不必引第三方库只取data:帧、跳过心跳、遇到[DONE]收尾。流式客户端必须readTimeout(0)关闭读取超时否则两帧间隔稍长就被掐断同时严禁 BODY 级日志拦截器它会先把整段流缓存进内存再输出流式退化成一次性。按 SSE 规范空行才是事件边界、多行data需拼接、:开头是注释而[DONE]之类的哨兵属服务端自定义接入前要核对文档。你们做流式对话时是怎么处理多气泡归位和断流重连的欢迎在评论区分享自己的做法。关键词标签#Android#流式对话#SSE#分层架构#MVVM#事件流#OkHttp#taskId