1. 为什么流式响应在 Android 上是个看起来简单、做起来全是坑的活如果你做过 Android 上的 AI 对话功能大概率经历过这样的场景接口在 Postman 里跑得好好的日志一行行往外吐但一搬到手机上要么半天不出一句话要么吐到一半突然卡死要么用户切个后台回来整个对话就断了。这不是你的代码写得差而是流式响应这套机制在移动端本身就比在服务端复杂得多——网络环境会变、系统会杀后台、UI 线程不能阻塞、用户还会随时点返回键。先把概念理清楚。所谓流式响应指的是服务端不把整段回答一次性返回而是把内容切成一个个小块chunk边生成边推送。用户看到的效果就是文字像打字一样一个个蹦出来而不是等三五秒后啪地出现一整段。这种体验对 AI 对话几乎是刚需因为大模型的生成本身就是逐 token 进行的如果强行等全部生成完再返回用户等待时间会非常难熬。而实现流式响应最常用的协议就是SSEServer-Sent Events。它的本质其实特别朴素一个长连接的 HTTP 请求服务端保持连接不关闭持续往响应体里写数据每条消息以data:开头以两个换行\n\n结尾。浏览器端有原生的EventSource可以直接用但 Android 上没有这个东西你得自己用 OkHttp 之类的客户端去读这个流然后手动解析。这就引出了本文要讲的三件事怎么在 Android 上把 SSE 流读下来并正确解析、流式链路里哪些地方最容易断、以及断了之后怎么优雅降级。我把它总结成三级降级的思路后面会详细展开。这套东西适合已经能跑通普通 HTTP 请求、但对流式处理还没什么经验的 Android 开发者也适合正在做 AI 对话类 App、被各种超时和断流折磨过的同学。需要提前说明的是本文涉及的具体参数比如超时时间、缓冲区大小都是基于常见实践给出的参考值你需要根据自己的服务端行为和网络环境做调整。核心是理解每个参数背后的意图而不是照抄数字。2. SSE 在 Android 上的真实工作方式从 OkHttp 到逐行解析2.1 为什么不能用普通的 Retrofit 接口来接收很多人第一反应是用 Retrofit 定义一个返回String或某个数据类的接口然后enqueue一个 Call。这条路走不通原因在于 Retrofit 默认会把整个响应体读完才回调给你而 SSE 的响应体是永不结束的直到服务端主动关闭。你等它读完就等于永远等不到。正确的做法是拿到原始的ResponseBody把它当成一个输入流来读。用 OkHttp 的话大致是这样val request Request.Builder() .url(chatUrl) .header(Accept, text/event-stream) .header(Content-Type, application/json) .post(jsonBody.toRequestBody(application/json.toMediaType())) .build() val client OkHttpClient.Builder() .readTimeout(0, TimeUnit.MILLISECONDS) // 关键读超时设为 0表示不超时 .connectTimeout(15, TimeUnit.SECONDS) .build() client.newCall(request).enqueue(object : Callback { override fun onResponse(call: Call, response: Response) { val source response.body?.source() ?: return while (!source.exhausted()) { val line source.readUtf8Line() ?: break // 处理这一行 } } override fun onFailure(call: Call, e: IOException) { /* 处理失败 */ } })这里有个极其关键的坑readTimeout必须设为 0 或者一个很大的值。OkHttp 默认的读超时是 10 秒意思是如果 10 秒内没有读到任何新数据就断开。但 AI 生成有时候会思考很久才吐第一个字或者两个 chunk 之间间隔超过 10 秒这时候连接就会被 OkHttp 自己掐断你会收到一个SocketTimeoutException日志里看起来就像流莫名其妙断了。我见过太多人卡在这里以为是服务端的问题其实是客户端超时配置没改。2.2 SSE 的报文格式与逐行解析逻辑SSE 的报文格式看起来简单但细节不少。一条完整的消息长这样data: {choices:[{delta:{content:你}}]} data: {choices:[{delta:{content:好}}]} data: [DONE]注意几个点每条data:后面跟的是内容消息之间用空行分隔流结束时会发一个特殊的结束标记OpenAI 风格是data: [DONE]。有些服务端还会发event:、id:、retry:这些字段但 AI 对话场景里通常只用data:。解析的时候不能假设一次 read 就是一条完整消息。TCP 是字节流一次读取可能拿到半条消息也可能拿到一条半。所以正确姿势是按行读取遇到空行就认为一条消息结束val dataBuffer StringBuilder() while (!source.exhausted()) { val line source.readUtf8Line() ?: break when { line.isEmpty() - { // 空行 一条消息结束 if (dataBuffer.isNotEmpty()) { handleMessage(dataBuffer.toString()) dataBuffer.clear() } } line.startsWith(data:) - { dataBuffer.append(line.removePrefix(data:).trim()) } // 其他字段event/id/retry按需处理 } }readUtf8Line()这个方法是 OkHttp 提供的它会一直读到换行符为止帮你处理了跨 buffer 的半行问题。如果你自己用InputStream.read()拼字符串很容易在 UTF-8 多字节字符被切断的地方出现乱码——中文尤其明显一个汉字三个字节正好卡在边界上就会变成问号。这是第二个高频坑。2.3 把解析结果安全地送到 UI 层解析出来的内容不能直接在网络线程里更新 UI。常规做法是用Flow或者LiveData把数据往上层抛。用 Kotlin 协程的话可以包一个callbackFlowfun streamChat(prompt: String): FlowString callbackFlow { val call client.newCall(buildRequest(prompt)) call.enqueue(object : Callback { override fun onResponse(call: Call, response: Response) { val source response.body?.source() ?: run { close(IOException(empty body)); return } try { val buffer StringBuilder() while (!source.exhausted()) { val line source.readUtf8Line() ?: break if (line.isEmpty()) { if (buffer.isNotEmpty()) { trySend(buffer.toString()) buffer.clear() } } else if (line.startsWith(data:)) { val payload line.removePrefix(data:).trim() if (payload [DONE]) { close(); return } buffer.append(payload) } } close() } catch (e: Exception) { close(e) } } override fun onFailure(call: Call, e: IOException) { close(e) } }) awaitClose { call.cancel() } }awaitClose里取消 call 很重要。用户如果中途退出页面Flow 被取消这个 call 必须跟着取消否则连接会一直挂着既浪费流量也占着线程。我踩过一次坑没写awaitClose结果用户快速进出对话页十几次后台堆了十几个没关闭的连接最后触发了系统的文件描述符上限整个 App 的网络请求全挂了。这个教训很深刻。3. 流式链路为什么会断把断流这件事拆开看3.1 移动网络下的三类典型断流流式响应断掉原因五花八门但归纳起来主要是三类。第一类是客户端主动超时。前面说的 OkHttpreadTimeout就是典型。还有一种是应用层自己设的整体超时比如你规定 30 秒内必须收到响应超时就 cancel但 AI 生成长回答可能超过 30 秒于是被你自己掐了。第二类是网络层中断。手机从 WiFi 切到 4G、进电梯信号丢失、地铁里频繁切换基站都会导致 TCP 连接断开。这种情况下你会收到IOException日志里常见的是unexpected end of stream或者stream disconnected before completion。这类断流客户端无能为力只能靠重连或降级。第三类是服务端或中间层主动关闭。比如网关有 idle timeout空闲超时如果两个 chunk 之间间隔太久网关认为这个连接没用了就关掉。你可能会看到idle timeout waiting for sse这样的报错。这种情况可以通过让服务端定期发送心跳比如每 15 秒发一个: keep-alive注释行来缓解但客户端也要做好应对准备。3.2 一个容易被忽略的细节切后台与进程优先级Android 系统对后台进程的管理很激进。用户把 App 切到后台如果内存紧张系统可能直接杀掉进程你的流式连接自然就断了。就算没被杀Doze 模式也会限制网络访问。这里有个实操经验不要指望流式连接能在后台稳定存活。正确的产品逻辑应该是——用户切后台时要么暂停接收保留已收到的内容要么把这次请求标记为可能中断等回到前台时检查状态。我见过有团队试图用前台 Service 硬扛结果被应用商店审核质疑为什么聊天要常驻前台服务得不偿失。如果确实需要在切后台后继续接收比如用户希望回答生成完能收到通知更稳妥的做法是让服务端把结果落库客户端回到前台后拉取完整结果而不是死磕那条流式连接。3.3 用日志把断流现场还原出来排查断流问题日志是第一手证据。建议在几个关键节点打点请求发出时、收到第一个 chunk 时、每收到 N 个 chunk 时、连接关闭时、发生异常时。异常里要打印e.message和e.cause因为 OkHttp 经常把真正的错误包在 cause 里。一个典型的排查链路是这样的用户反馈回答到一半停了→ 查日志发现最后一条是SocketTimeoutException: timeout→ 看时间戳距离上一个 chunk 正好 10 秒 → 确认是readTimeout没改 → 改成 0 后问题消失。整个过程如果没日志你只能靠猜。提示线上环境建议把流式相关的日志做采样上报不要全量打否则日志量会爆炸。但异常日志要全量收集这是定位问题的命脉。4. 三级降级让对话在断流时依然能用4.1 降级思路的整体设计所谓三级降级核心思想是当理想方案流式不可用时逐级退到次优方案保证功能始终可用而不是直接报错。这三级的划分是这样的级别方案触发条件用户体验一级正常流式连接正常逐字输出最佳体验二级流式重连 断点续传中途断流短暂停顿后继续输出三级非流式一次性请求流式多次失败等待后整段出现这个设计的关键在于每一级都要有明确的触发条件和退出机制不能无限重试。我的经验是一级到二级允许重试 1 次二级到三级直接切换三级失败才向用户报错。重试次数太多会让用户等得心焦还不如早点给个完整结果。4.2 二级降级断点续传怎么实现断点续传的前提是服务端支持。常见做法是客户端记录已经收到的内容长度或最后一个 chunk 的序号重连时把这个信息带上服务端从断点继续生成。如果服务端不支持那就只能退而求其次——重连后重新请求但把已收到的内容先展示出来新内容追加在后面可能重复需要去重。实现上重连逻辑要包一层suspend fun streamWithRetry(prompt: String, maxRetry: Int 1): FlowString flow { var attempt 0 var received StringBuilder() while (attempt maxRetry) { try { streamChat(prompt, received.length).collect { chunk - received.append(chunk) emit(chunk) } returnflow // 正常结束 } catch (e: Exception) { attempt if (attempt maxRetry) throw e delay(500) // 稍等再重连避免立刻又撞上同一个问题 } } }这里的delay(500)不是随便写的。断流后立刻重连很可能撞上服务端还没释放资源、或者网络还没恢复的情况稍等一下成功率更高。但也不能等太久否则用户会觉得卡住了。4.3 三级降级非流式兜底的价值非流式请求就是最普通的 POST等全部生成完一次性返回。它的缺点是慢但优点是极其稳定——没有长连接、没有超时困扰、没有解析复杂度。当流式反复失败时切到非流式往往能救回来。这里有个产品层面的取舍非流式请求的等待时间可能长达十几秒用户会以为 App 卡死了。所以切到三级降级时一定要给明确的加载提示比如正在生成完整回答请稍候而不是让界面一片空白。我见过有 App 在降级时什么都不显示用户以为死机了直接卸载非常可惜。另外三级降级的超时时间要设得比流式长很多比如 60 秒甚至 120 秒因为大模型生成完整回答本来就慢。但也不能无限等超过阈值还是要报错让用户重试。5. 那些文档里不会写的实操细节5.1 缓冲区与背压别让 UI 被 chunk 淹没流式响应来得快的时候一秒可能来几十个 chunk。如果每个 chunk 都触发一次 UI 更新主线程会被频繁的 recomposition 或 notifyItemChanged 拖垮界面反而卡顿。解决办法是做批量合并攒够一定数量或者间隔一定时间比如 50ms再统一刷新一次 UI。// 用 Flow 的 buffer sample 做节流 streamChat(prompt) .buffer(Channel.UNLIMITED) .sample(50) // 每 50ms 取最新值 .collect { updateUI(it) }sample会丢弃中间值只保留最新的对于文字追加这种场景刚好合适——反正最终显示的是完整文本中间过程丢几帧无所谓。但要注意如果服务端返回的是结构化数据比如带工具调用就不能随便丢得用buffer保证不丢数据。5.2 中文乱码与 UTF-8 边界问题前面提过中文在流式解析里特别容易出乱码。根本原因是 UTF-8 是变长编码一个汉字占 3 字节如果一次读取正好把某个汉字切成两半直接按字节转字符串就会得到乱码。readUtf8Line()内部处理了这个问题但如果你自己用InputStream读一定要用InputStreamReader指定 UTF-8或者干脆用 OkHttp 的source。还有一个隐蔽的坑有些服务端在 chunk 边界处会插入换行导致你以为一条消息结束了其实只是内容里本来就有换行。所以解析时要以空行为消息分隔符而不是任意换行。这个区别在返回 Markdown 格式内容时尤其重要因为 Markdown 里换行很常见。5.3 用户主动取消时的资源清理用户点停止生成或者退出页面必须把连接干净地关掉。除了前面说的awaitClose里 cancel call还要注意协程作用域的取消。如果用的是viewModelScope页面销毁时协程自动取消但 OkHttp 的 call 不会自动取消必须手动处理。我建议把取消做成一个显式的操作而不是依赖生命周期。因为用户可能只是想停止当前这次生成但页面还在这时候如果依赖onDestroy才取消连接会一直挂着。显式取消的代码大概是这样private var currentCall: Call? null fun stopGenerating() { currentCall?.cancel() currentCall null }5.4 心跳与保活让连接别被中间层掐掉如果你的服务端前面有网关或负载均衡它们通常有 idle timeout。客户端能做的是配合服务端发心跳。SSE 里心跳通常是一个以冒号开头的注释行比如: ping客户端解析时直接忽略即可。但要注意心跳行也会触发readUtf8Line()返回所以你的解析逻辑要能识别并跳过它别把它当成数据。if (line.startsWith(:)) { // 心跳或注释忽略 returnloop }如果服务端不发心跳你也可以在客户端设一个静默计时器超过一定时间没收到任何数据就主动重连。这个阈值建议设得比网关的 idle timeout 小一点比如网关是 60 秒你就设 45 秒。6. 从一次真实断流事故看排查与修复的完整链路说个我亲身经历的事故。某次上线后用户反馈回答经常只出一半就停了而且集中在特定运营商网络下。我们按下面的顺序排查第一步看客户端日志。发现断流时抛的是IOException: unexpected end of stream不是超时。这说明不是readTimeout的问题而是连接被对端关了。第二步看断流的时间点。统计发现大部分断流发生在请求发出后 30 秒左右。这个数字很可疑像是某个中间层的超时。第三步抓包对比。在 WiFi 下复现不了在特定运营商 4G 下能复现。抓包发现运营商侧有个透明代理对超过 30 秒没有完整响应的连接会强制关闭。第四步验证方案。让服务端每 15 秒发一次心跳客户端忽略心跳行。改完后同样网络下断流率从 30% 降到 2% 以下。第五步加兜底。即使有心跳还是会有零星断流于是把二级降级重连和三级降级非流式都补上确保极端情况下功能可用。这个事故给我的最大启发是流式链路的稳定性是客户端、服务端、中间网络三方共同决定的任何一方的问题都会表现为客户端断流。所以排查时不能只盯着客户端代码要学会从时间规律、网络环境、对端行为多个角度去推断。7. 关于参数配置和工程化的一些个人经验最后聊几个工程化层面的东西这些是我做了几个 AI 对话项目后攒下来的。超时参数要分层设置。连接超时connectTimeout可以短一点15 秒足够读超时readTimeout对流式请求必须设 0 或很大整体超时用协程的withTimeout要单独控制比如 120 秒。这三者职责不同别混在一起。重试要有退避。断流后立刻重试容易连续失败建议第一次等 500ms第二次等 1s最多重试 1-2 次。重试次数和间隔最好做成可配置的方便线上调整。降级要可观测。每次触发降级都上报一个埋点统计各级降级的触发频率。如果三级降级频繁触发说明流式链路有系统性问题需要重点优化。这个数据比任何主观感受都可靠。测试要覆盖弱网。用 Android Studio 的网络限速功能或者真机在电梯、地铁里测。很多断流问题只有在真实弱网下才暴露模拟器上跑得再顺也没用。别忽略服务端配合。客户端能做的降级是有限的真正稳定的流式体验需要服务端支持心跳、断点续传、合理的超时配置。如果服务端不配合客户端再怎么优化也是治标不治本。推动服务端一起改比在客户端堆补丁有效得多。这套流式响应加三级降级的方案我在两个项目里落地过整体断流导致的用户投诉下降了八成以上。核心不是某个技巧多高明而是把会断当成常态来设计而不是假设它永远不断。想清楚每一级降级的触发条件和用户体验剩下的就是把这些逻辑老老实实写进代码里。
