企业官网建设流程全解析

OkHttp Server-Sent Events（SSE）模块实战：EventSource 事件流接入指南

【免费下载链接】okhttpA meticulous HTTP client for the JVM, Android, and GraalVM.项目地址: https://gitcode.com/gh_mirrors/okh/okhttp

Server-Sent Events（SSE）是服务端通过 HTTP 长连接向客户端单向推送事件的标准协议，适合实时通知、进度推送、行情更新等场景。OkHttp 在独立模块okhttp-sse中提供了对 SSE 的实验性支持，本文基于仓库中 okhttp-sse/README.md 及其源码，系统讲解如何通过 OkHttp 的EventSourceAPI 订阅事件流、解析标准 SSE 帧，并处理重连、鉴权与超时等工程问题。读完本文，你将掌握 SSE 流式推送在 JVM / Android 应用中的完整接入方案。

模块概览与依赖引入

okhttp-sse是 OkHttp 官方仓库中独立发布的实验性模块（对应 okhttp-sse/Module.md 中 "Support for server-sent events" 的描述）。需要注意：该 API 尚不稳定，随时可能变更（README 明确标注 "Experimental support for server-sent events. API is not considered stable and may change at any time."）。

添加依赖

在 Gradle 项目中，通过testImplementation引入依赖（当前仓库版本为 5.5.0）：

testImplementation("com.squareup.okhttp3:okhttp-sse:5.5.0")

注意：README 中该依赖声明在testImplementation配置下，实际用于生产环境时应按项目需要改为implementation。该模块以okhttp3为核心依赖，JPMS 模块声明见 okhttp-sse/src/main/java9/module-info.java：模块名okhttp3.sse，requires okhttp3并exports okhttp3.sse，因此 Java 9+ 模块化项目可直接引用。

核心 API 全景

okhttp-sse对外只暴露三个类，全部位于okhttp3.sse包，二进制 API 定义见 okhttp-sse/api/okhttp-sse.api。

1. EventSource：事件源句柄

EventSource.kt 定义了两个方法：

• request(): Request—— 返回发起该事件源的原始请求；
• cancel()—— "Immediately and violently release resources"（立即且彻底地释放该事件源占用的资源）；若事件源已关闭或已取消，此操作无副作用。

其内部接口EventSource.Factory是创建事件源的入口：

fun interface Factory { fun newEventSource( request: Request, listener: EventSourceListener, ): EventSource }

创建事件源即会发起异步连接流程，连接成功或失败后listener会收到通知；调用方在不再使用时必须主动 cancel返回的事件源。

2. EventSourceListener：事件回调

EventSourceListener.kt 是一个抽象类，四个回调方法均带默认空实现，可按需覆写：

3. EventSources：工厂与工具入口

EventSources.kt 是object单例，提供两个@JvmStatic方法：

• createFactory(callFactory: Call.Factory): EventSource.Factory—— 由OkHttpClient（其实现了Call.Factory）创建事件源工厂；
• processResponse(response, listener)—— 在已有 OkHttpResponse之上直接挂接 SSE 处理（用于自行管理 Call 的场景）。

createFactory内部有一个关键细节：如果请求头中没有显式设置Accept，会自动补上Accept: text/event-stream；若已设置则保留原值。这一行为有测试覆盖（见下节）。

最小可运行示例

结合 EventSourceHttpTest.kt 中的用法，一个完整的接入流程如下：

import okhttp3.OkHttpClient import okhttp3.Request import okhttp3.sse.EventSource import okhttp3.sse.EventSourceListener import okhttp3.sse.EventSources.createFactory import okhttp3.Response val client = OkHttpClient() // 1. 创建事件源工厂 val factory = createFactory(client) // 2. 定义监听器 val listener = object : EventSourceListener() { override fun onOpen(eventSource: EventSource, response: Response) { println("连接已建立") } override fun onEvent( eventSource: EventSource, id: String?, type: String?, data: String, ) { println("收到事件: id=$id, type=$type, data=$data") } override fun onClosed(eventSource: EventSource) { println("连接已关闭") } override fun onFailure( eventSource: EventSource, t: Throwable?, response: Response?, ) { println("连接失败: $t") } } // 3. 发起请求并订阅 val request = Request.Builder() .url("https://example.com/events") .build() val eventSource: EventSource = factory.newEventSource(request, listener) // 4. 不再使用时释放资源 // eventSource.cancel()

事件解析协议详解

SSE 帧的解析逻辑在 ServerSentEventReader.kt 中实现。它用 Okio 的Options一次性匹配 20 种前缀模式（\r\n、\r、\n三种行结束符分别与data、id、event、retry字段组合），支持：

• 多行 data：多条data:行会被累积，并以换行符拼接。测试multiline用例验证了data: YHOO、data: +2、data: 10解析为"YHOO\n+2\n10"（见 ServerSentEventIteratorTest.kt）；
• event 类型：event: add指定事件类型，随onEvent的type参数返回；
• id 与重连：id:行更新lastId并在事件携带；单独的id行（无冒号值）会清空 id；
• retry 指令：retry:解析为毫秒数，通过onRetryChange回调通知——但 RealEventSource.kt 中明确忽略该值，不做自动重连（注释 "Ignored. We do not auto-retry."）；
• 注释与空行：以:开头的注释行被跳过；data为空的帧不会触发onEvent（completeEvent中data.size == 0L时直接返回）。

processNextEvent()每次处理一条事件，EOF 时返回false，驱动整个读取循环。

底层工作流程：RealEventSource

RealEventSource.kt 同时实现了EventSource、ServerSentEventReader.Callback与 OkHttp 的Callback，核心流程如下：

1. connect：callFactory.newCall(request).enqueue(this)异步发起请求；
2. onResponse 校验：响应不成功（!isSuccessful）或Content-Type不是text/event-stream（校验逻辑见isEventStream()：要求type == "text" && subtype == "event-stream"）时，直接回调onFailure；
3. 取消全量超时：SSE 是长连接，call?.timeout()?.cancel()取消整次调用的超时定时器，避免长连接被误杀；
4. 剥离响应体：response.stripBody()替换 body，保证外部回调无法读到真实流数据；
5. 读取循环：listener.onOpen后，在while (!canceled && reader.processNextEvent())中持续消费事件；
6. 收尾：读取异常时回调onFailure；被取消时回调onFailure(IOException("canceled"))；正常读到 EOF 时回调onClosed。

因此事件回调线程与 OkHttp 异步回调线程一致（即 OkHttp 的 Dispatcher 线程），开发者不应在onEvent中执行耗时操作。

关键边界行为与测试验证

仓库测试 EventSourceHttpTest.kt 覆盖了以下工程要点：

其中sseReauths证明：SSE 连接同样走 OkHttp 的拦截器与重试链路——只要配置了Authenticator，401 响应会自动触发携带凭据的重试，无需手动处理；而没有Authenticator时 401 直接导致onFailure（sseWithoutAuthenticator）。测试还通过EventRecorder验证了一次完整 SSE 调用会依次触发CallStart→ … →ResponseBodyStart→ResponseBodyEnd→CallEnd等标准 CallEvent 事件（见eventListenerEvents），说明 SSE 完全复用 OkHttp 的调用事件体系，便于埋点观测。

使用建议与注意事项

基于源码与测试，给出如下工程建议：

1. 必须调用cancel()：事件源不再使用时（如页面销毁、任务取消），务必调用EventSource.cancel()以释放连接资源；取消路径会在onFailure中以IOException("canceled")收尾。
2. 主动处理重连：模块不做自动重连（retry指令被忽略），服务端断流后应自行决定退避策略并重新newEventSource。
3. 校验 Content-Type：服务端必须返回text/event-stream，否则会收到Invalid content-type失败回调；可通过Authenticator/拦截器链路处理鉴权 401 重试。
4. 注意 API 不稳定：作为实验性模块，升级 OkHttp 版本时需留意EventSource相关 API 的变更。
5. 长连接与超时：连接建立后全量调用超时会被取消，但读取数据仍受底层 socket 读超时约束，可结合 OkHttpClient 的 read 超时配置。

相关资源

• 模块说明：okhttp-sse/README.md、okhttp-sse/Module.md
• 对外 API：okhttp-sse/api/okhttp-sse.api
• 核心实现：EventSource.kt、EventSourceListener.kt、EventSources.kt、RealEventSource.kt、ServerSentEventReader.kt
• 测试用例：EventSourceHttpTest.kt、ServerSentEventIteratorTest.kt

【免费下载链接】okhttpA meticulous HTTP client for the JVM, Android, and GraalVM.项目地址: https://gitcode.com/gh_mirrors/okh/okhttp