本文面向 Android/Kotlin 开发者，目标是从基础概念到工程落地，系统理解并实践：
先用一个故事把这篇笔记的顺序串起来。
很久以前，Web 的主要用途就是浏览文档。
用户点一个链接，浏览器发一个 HTTP 请求：
浏览器 -> 给我 index.html
服务器 -> 好，给你 index.html
这时服务器很轻松。
因为每次请求都是独立的。
服务器不需要记住：
这就是 HTTP 的一个重要特点：无状态。
无状态不是缺点，而是早期 Web 的简单性来源。
HTTP 无状态 = 每个请求对服务器来说都像第一次见面。
后来 Web 应用复杂了。
出现了登录、购物车、订单、个人中心。
服务器突然发现：
我不能每次都假装不认识你了。
我必须知道谁登录了，谁把商品放进了购物车。
可是 HTTP 本身是无状态的。
怎么办？
一个经典方案是 session + cookie。
流程大概是：
1. 用户登录。
2. 服务器创建一份 session，里面记录 userId、登录态、权限等。
3. 服务器生成一个 sessionId。
4. 服务器通过 Set-Cookie 把 sessionId 发给客户端。
5. 客户端后续请求自动带上 Cookie: sessionId=xxx。
6. 服务器根据 sessionId 查 session，就知道你是谁。
形象理解：
sessionId 就像酒店房卡。
客户端拿着房卡。
服务器前台保存着房卡对应的住客信息。
你每次来，出示房卡，前台就知道你是谁。
session 方案能工作，但服务器开始不轻松了。
如果有 10 万、100 万用户在线，服务器就要维护大量 session。
单机还好，集群会更麻烦。
例如：
用户在机器 A 登录，session 存在机器 A。
下一次请求被负载均衡打到机器 B。
机器 B 没有这个 session。
机器 B 不认识这个用户。
常见补救方案有：
这些方案都能解决问题，但也都引入新成本。
服务器会想：
能不能不要让我保存这么多 session？
能不能让客户端自己保存登录凭证，我只负责验证？
token 的思路是：
服务器登录时发给客户端一个 token。
客户端后续请求带上 token。
服务器不一定保存 token，只需要验证 token 是否可信。
如果是 JWT 这类签名 token，里面可能包含：
服务器验证 token 时：
1. 拿到 token。
2. 验证签名是否正确。
3. 验证是否过期。
4. 从 token 中取出 userId。
5. 判断用户权限。
形象理解：
session 像酒店前台保存房客资料。
token 像带防伪签名的通行证。
服务器不一定保存每张通行证，只需要验证防伪签名。
但 token 也有风险。
如果 token 被偷，攻击者就可以冒充用户。
所以：
登录问题解决后，新的问题来了。
有些业务不只是“我请求，你响应”。
例如：
这些业务需要服务端主动通知客户端。
普通 HTTP 的模式是：
客户端问一次，服务端答一次。
客户端不问，服务端通常没法主动说话。
于是出现了几种实时方案：
| 方案 | 做法 | 问题 |
|---|---|---|
| 短轮询 | 客户端每隔几秒请求一次 | 浪费请求，实时性一般 |
| 长轮询 | 服务端先挂住请求，有消息再返回 | 比短轮询好，但连接频繁重建 |
| SSE | 服务端单向推送 | 主要适合服务端到客户端单向流 |
| WebSocket | 建立双向长连接 | 更适合双向实时通信 |
WebSocket 就像把“每次都打电话问一下”变成“电话一直通着”。
HTTP 轮询 = 每隔几秒打一次电话问：有消息吗？
WebSocket = 电话接通后不挂，谁有事谁直接说。
WebSocket 建立长连接后，客户端和服务端都可以主动发消息。
但移动端网络不稳定。
可能发生：
这时需要心跳和重连。
心跳负责确认连接是否还活着：
客户端定期 ping。
服务端回复 pong。
如果连续几次没有 pong，说明连接可能坏了。
重连负责恢复连接：
发现连接坏了。
关闭旧连接。
等待一段时间。
重新建立 WebSocket。
重新鉴权。
重新订阅。
拉取断线期间缺失消息。
在 Android 项目里，常见分工是：
Retrofit：负责普通 HTTP API，例如登录、刷新 token、拉历史消息。
OkHttp WebSocket：负责真正的 WebSocket 长连接。
Gson：负责 JSON 字符串和 Kotlin 对象互转。
协程 Flow / RxJava：负责把异步事件组织成数据流。
Repository：负责把底层连接事件变成业务状态。
ViewModel：负责给 UI 暴露状态。
UI：负责展示连接状态和消息。
一个完整 App 的典型顺序是：
1. Retrofit 调登录接口，拿到 token。
2. OkHttp newWebSocket，握手 Header 带 token。
3. WebSocket onOpen，连接建立成功。
4. 如果服务端要求，发送 auth 业务消息。
5. 启动心跳。
6. 收到 JSON 文本消息。
7. Gson 解析成 Kotlin data class。
8. 根据 type 分发到业务模块。
9. 网络断开时触发重连。
10. 重连成功后重新鉴权、重新订阅。
11. Retrofit 拉断线期间缺失的历史消息。
这就是本文后续章节的组织顺序。
建议按下面顺序学习：
第一层：为什么需要 WebSocket
第二层：WebSocket 基础协议和 ws/wss
第三层：Android 工程依赖和整体架构
第四层：Retrofit 负责 HTTP 登录和 token
第五层：OkHttp newWebSocket 建立长连接
第六层：Gson 解析 JSON 业务协议
第七层：封装 WebSocketClient
第八层：线程、生命周期、ViewModel/Repository
第九层：心跳、重连、网络变化
第十层：鉴权、TLS、安全
第十一层：协程 Flow / RxJava 事件流封装
第十二层：ACK、补偿、去重、订阅恢复
第十三层：测试、排错、性能、上线检查
整体架构图：
Activity / Fragment
|
v
ViewModel
|
v
WebSocketRepository
|
|-------------------------|
v v
AndroidWebSocketClient Retrofit ApiService
| |
v v
OkHttp WebSocket OkHttp HTTP Call
| |
v v
WebSocket Server HTTP Server
核心分工：
| 层 | 负责什么 |
|---|---|
| UI | 展示连接状态、消息、错误 |
| ViewModel | 维护页面状态，调用 Repository |
| Repository | 组合 HTTP、WebSocket、缓存、业务规则 |
| WebSocketClient | 管理连接、发送、关闭、重连、心跳 |
| Retrofit ApiService | 登录、刷新 token、拉历史消息、上传日志 |
| Gson Parser | JSON 到 Kotlin 对象的转换 |
| OkHttp | 真正执行 HTTP 请求和 WebSocket 连接 |
TCP 是传输层协议，提供可靠的字节流传输。
WebSocket 底层依赖 TCP。
TCP 负责：
但 TCP 只传输字节流，不理解 JSON，也不理解业务消息。
如果直接基于 TCP 写业务协议，需要自己处理粘包、拆包、消息边界。
WebSocket 已经在 TCP 之上定义了帧格式，所以 OkHttp 的 onMessage(text) 通常已经是一条完整 WebSocket 消息。
HTTP 是请求响应协议。
一次典型 HTTP 交互：
客户端 -> HTTP Request -> 服务端
客户端 <- HTTP Response <- 服务端
HTTP 本身无状态。
也就是说，协议本身不会自动记住上一次请求是谁发的。
登录态通常靠 cookie/session 或 token 来补足。
WebSocket 建连时，先发起一个 HTTP 请求，请求服务端把协议升级为 WebSocket。
简化握手请求：
GET /socket HTTP/1.1
Host: api.example.com
Connection: Upgrade
Upgrade: websocket
Sec-WebSocket-Key: xxxxx
Sec-WebSocket-Version: 13
Authorization: Bearer <token>
服务端同意升级：
HTTP/1.1 101 Switching Protocols
Connection: Upgrade
Upgrade: websocket
Sec-WebSocket-Accept: xxxxx
101 Switching Protocols 表示握手成功。
握手成功后，连接进入 WebSocket 帧通信阶段。
| Scheme | 含义 | 类比 |
|---|---|---|
ws:// |
明文 WebSocket | 类似 http:// |
wss:// |
TLS 加密 WebSocket | 类似 https:// |
生产环境优先使用 wss://。
原因：
长连接不是永远不断开的连接。
更准确地说：
长连接 = 建立后尽量保持并复用同一条连接，持续收发数据。
优点：
成本：
全双工表示双方可以同时发送和接收。
WebSocket 建立后：
这和普通 HTTP 的请求响应模型不同。
WebSocket 传输的是帧。
常见帧类型：
| 类型 | 说明 | OkHttp 表现 |
|---|---|---|
| Text | 文本消息，常见内容是 JSON | onMessage(webSocket, text) |
| Binary | 二进制消息 | onMessage(webSocket, bytes) |
| Ping | 协议层心跳探测 | OkHttp pingInterval 可自动发送 |
| Pong | 对 Ping 的响应 | 多数由框架处理 |
| Close | 关闭帧 | onClosing、onClosed |
业务中最常见的是 Text 帧。
半开连接指的是：
客户端以为连接还活着，但服务端、中间代理或 NAT 已经断开了连接。
半开连接很常见。
原因可能是：
心跳机制主要就是为了发现半开连接。
心跳是长连接中的定期探测机制。
定义：
心跳 = 定期发送一个很小的数据包，用来确认连接还活着，并防止连接长时间空闲被清理。
心跳解决三件事：
| 问题 | 作用 |
|---|---|
| 连接是否还活着 | ping 后等 pong |
| 中间设备是否清理空闲连接 | 定期产生少量流量 |
| 业务在线状态是否可信 | 多次失败后标记离线并重连 |
ACK 是 acknowledgement 的缩写，表示确认收到。
webSocket.send() 返回 true 只表示消息进入 OkHttp 发送队列。
它不代表：
如果业务需要可靠性，就要让服务端回 ACK。
客户端发送：
{"type":"chat_send","requestId":"req_1","data":{"content":"hello"}}
服务端回复：
{"type":"ack","requestId":"req_1","code":0,"message":"ok"}
序列化：
Kotlin 对象 -> JSON 字符串
反序列化：
JSON 字符串 -> Kotlin 对象
Gson 负责这两件事。
WebSocket 发送前通常要序列化。
WebSocket 接收后通常要反序列化。
基础依赖：
dependencies {
implementation("com.squareup.okhttp3:okhttp:4.12.0")
implementation("com.google.code.gson:gson:2.10.1")
}
如果项目使用 Retrofit、协程、RxJava：
dependencies {
implementation("com.squareup.okhttp3:okhttp:4.12.0")
implementation("com.squareup.okhttp3:logging-interceptor:4.12.0")
implementation("com.squareup.retrofit2:retrofit:2.11.0")
implementation("com.squareup.retrofit2:converter-gson:2.11.0")
implementation("com.squareup.retrofit2:adapter-rxjava3:2.11.0")
implementation("com.google.code.gson:gson:2.10.1")
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.1")
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-android:1.8.1")
implementation("io.reactivex.rxjava3:rxjava:3.1.8")
implementation("io.reactivex.rxjava3:rxandroid:3.0.2")
}
分工：
| 库 | 作用 |
|---|---|
| OkHttp | HTTP 和 WebSocket 底层网络库 |
| Retrofit | 声明式 HTTP API，例如登录、刷新 token、拉历史消息 |
| Gson | JSON 和 Kotlin 对象互转 |
| Coroutines | Kotlin 官方异步模型，可以用 Flow 包装事件 |
| RxJava | 响应式事件流，老项目常见 |
网络权限：
<uses-permission android:name="android.permission.INTERNET" />
监听网络变化：
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
如果测试环境使用 ws:// 明文连接，Android 9 及以上可能需要网络安全配置。
生产环境应优先使用 wss://。
WebSocket 通常不是 App 第一个请求。
常见流程是：
1. App 用 Retrofit 调登录接口。
2. 服务端返回 access token 和 refresh token。
3. App 用 token 建立 WebSocket。
4. WebSocket 断线后，App 可能用 Retrofit 拉历史消息。
5. token 过期后，App 用 Retrofit 刷新 token。
所以 Retrofit 和 WebSocket 是配合关系。
Retrofit 负责普通 HTTP API。
OkHttp newWebSocket 负责 WebSocket 长连接。
data class LoginRequest(
val username: String,
val password: String
)
data class LoginResponse(
val accessToken: String,
val refreshToken: String,
val expiresIn: Long
)
data class RefreshTokenRequest(
val refreshToken: String
)
data class HistoryMessageResponse(
val messages: List<ChatMessagePayload>,
val nextCursor: String?
)
interface ApiService {
@POST("auth/login")
suspend fun login(@Body request: LoginRequest): LoginResponse
@POST("auth/refresh")
suspend fun refreshToken(@Body request: RefreshTokenRequest): LoginResponse
@GET("chat/history")
suspend fun getChatHistory(
@Query("roomId") roomId: String,
@Query("cursor") cursor: String?
): HistoryMessageResponse
}
创建 Retrofit：
val okHttpClient = OkHttpClient.Builder()
.connectTimeout(10, TimeUnit.SECONDS)
.writeTimeout(10, TimeUnit.SECONDS)
.addInterceptor { chain ->
val request = chain.request().newBuilder()
.addHeader("X-Platform", "android")
.addHeader("X-App-Version", "1.0.0")
.build()
chain.proceed(request)
}
.build()
val retrofit = Retrofit.Builder()
.baseUrl("https://api.example.com/")
.client(okHttpClient)
.addConverterFactory(GsonConverterFactory.create())
.build()
val apiService = retrofit.create(ApiService::class.java)
class AuthRepository(
private val apiService: ApiService
) {
private var accessToken: String? = null
private var refreshToken: String? = null
suspend fun login(username: String, password: String) {
val response = apiService.login(
LoginRequest(username = username, password = password)
)
accessToken = response.accessToken
refreshToken = response.refreshToken
}
fun currentAccessToken(): String? {
return accessToken
}
suspend fun refreshAccessToken(): Boolean {
val currentRefreshToken = refreshToken ?: return false
return runCatching {
val response = apiService.refreshToken(
RefreshTokenRequest(currentRefreshToken)
)
accessToken = response.accessToken
refreshToken = response.refreshToken
}.isSuccess
}
}
注意：
协程里可以用 Mutex 防止并发刷新：
private val refreshMutex = Mutex()
suspend fun refreshAccessTokenSafely(): Boolean {
return refreshMutex.withLock {
authRepository.refreshAccessToken()
}
}
val webSocketOkHttpClient = okHttpClient.newBuilder()
.readTimeout(0, TimeUnit.MILLISECONDS)
.pingInterval(20, TimeUnit.SECONDS)
.build()
val request = Request.Builder()
.url("wss://api.example.com/socket")
.addHeader("Authorization", "Bearer $token")
.build()
val listener = object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) {
webSocket.send("{\"type\":\"hello\"}")
}
override fun onMessage(webSocket: WebSocket, text: String) {
println("receive: $text")
}
override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
println("failure: ${t.message}, code=${response?.code}")
}
override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
println("closed: code=$code, reason=$reason")
}
}
val webSocket = webSocketOkHttpClient.newWebSocket(request, listener)
这个示例只适合入门。
实际项目还需要：
val webSocketClient = OkHttpClient.Builder()
.connectTimeout(10, TimeUnit.SECONDS)
.readTimeout(0, TimeUnit.MILLISECONDS)
.writeTimeout(10, TimeUnit.SECONDS)
.pingInterval(20, TimeUnit.SECONDS)
.build()
说明：
| 配置 | 说明 |
|---|---|
connectTimeout |
TCP 建连超时 |
readTimeout |
WebSocket 长连接通常设为 0，避免普通读超时干扰 |
writeTimeout |
写数据超时 |
pingInterval |
OkHttp 自动发送 WebSocket 协议层 ping |
如果 Retrofit 和 WebSocket 共用 OkHttpClient，可以用 newBuilder() 派生 WebSocket 专用配置。
Request 是 WebSocket 的握手请求。
Header、token、cookie、query 参数都在这里配置。
val request = Request.Builder()
.url("wss://api.example.com/socket?deviceId=$deviceId")
.addHeader("Authorization", "Bearer $token")
.addHeader("X-App-Version", appVersion)
.build()
注意：
常用方法：
| 方法 | 说明 |
|---|---|
send(text) |
发送文本消息 |
send(bytes) |
发送二进制消息 |
close(code, reason) |
优雅关闭 |
cancel() |
立即取消 |
queueSize() |
当前待发送队列大小 |
send 返回 Boolean。
true 只代表进入发送队列，不代表服务端收到。
核心回调：
| 回调 | 触发时机 |
|---|---|
onOpen |
HTTP Upgrade 成功，连接打开 |
onMessage(String) |
收到文本消息 |
onMessage(ByteString) |
收到二进制消息 |
onClosing |
收到关闭帧，连接正在关闭 |
onClosed |
连接已经关闭 |
onFailure |
握手失败、网络失败、读写失败、异常断开 |
通常不需要调用 super.onXxx()，OkHttp 默认实现基本是空实现。
成功流程：
newWebSocket
-> onOpen
-> onMessage
-> onMessage
-> onClosing
-> onClosed
失败流程：
newWebSocket
-> onFailure
已连接后异常断开：
onOpen
-> onMessage
-> 网络断开
-> onFailure
onFailure 中的 response?.code 很重要。
例如：
401：token 无效或过期。403：无权限。404：WebSocket 路径错误。500：服务端错误。null：可能是网络异常、TLS 异常、IO 异常。WebSocket 只规定连接和帧，不规定 JSON 字段。
所以必须和服务端约定业务协议。
推荐统一消息信封 envelope：
{
"type": "chat_message",
"requestId": "req_10001",
"code": 0,
"message": "ok",
"data": {
"roomId": "room_1",
"senderId": "u_1",
"content": "hello"
},
"timestamp": 1710000000000
}
字段含义：
| 字段 | 说明 |
|---|---|
type |
消息类型，用于分发 |
requestId |
请求 ID，用于匹配响应 |
code |
业务响应码 |
message |
业务错误或提示信息 |
data |
业务数据，不同 type 结构不同 |
timestamp |
时间戳 |
对象转 JSON：
val gson = Gson()
val payload = ChatSendPayload(
roomId = "room_1",
content = "hello"
)
val json = gson.toJson(payload)
JSON 转对象：
val payload = gson.fromJson(json, ChatMessagePayload::class.java)
字段名不一致时使用 @SerializedName：
data class UserDto(
@SerializedName("user_id")
val userId: String?,
@SerializedName("user_name")
val userName: String?
)
Gson 是 Java 反射库，不完全理解 Kotlin 的非空语义。
如果服务端缺字段，非空字段也可能被解析成 null。
所以网络 DTO 建议：
示例：
data class ChatMessagePayload(
val roomId: String?,
val senderId: String?,
val content: String?,
val messageId: String?
)
data class WsEnvelope(
@SerializedName("type")
val type: String?,
@SerializedName("requestId")
val requestId: String? = null,
@SerializedName("data")
val data: JsonElement? = null,
@SerializedName("code")
val code: Int? = null,
@SerializedName("message")
val message: String? = null,
@SerializedName("timestamp")
val timestamp: Long? = null
)
data 使用 JsonElement，因为不同 type 的 data 结构不同。
解析顺序：
先解析外层 WsEnvelope。
根据 envelope.type 判断消息类型。
再把 envelope.data 解析成对应 data class。
sealed class ServerMessage {
data class AuthOk(val payload: AuthOkPayload) : ServerMessage()
data class ChatMessage(val payload: ChatMessagePayload) : ServerMessage()
data class SystemNotice(val payload: SystemNoticePayload) : ServerMessage()
data class Pong(val payload: PongPayload?) : ServerMessage()
data class Ack(val payload: AckPayload) : ServerMessage()
data class BusinessError(val code: Int?, val message: String?, val requestId: String?) : ServerMessage()
data class Unknown(val type: String, val rawData: JsonElement?) : ServerMessage()
}
data class AuthOkPayload(
val userId: String?,
val sessionId: String?
)
data class ChatMessagePayload(
val roomId: String?,
val senderId: String?,
val content: String?,
val messageId: String?
)
data class SystemNoticePayload(
val title: String?,
val content: String?
)
data class PongPayload(
val serverTime: Long?
)
data class AckPayload(
val requestId: String?,
val code: Int?,
val message: String?
)
class WsMessageParser(
private val gson: Gson = Gson()
) {
sealed class ParseResult {
data class Success(val message: ServerMessage) : ParseResult()
data class Failure(val raw: String, val error: Throwable) : ParseResult()
}
fun parse(raw: String): ParseResult {
return try {
val envelope = gson.fromJson(raw, WsEnvelope::class.java)
val type = envelope.type
if (type.isNullOrBlank()) {
return ParseResult.Failure(
raw = raw,
error = IllegalArgumentException("missing message type")
)
}
val message = when (type) {
"auth_ok" -> ServerMessage.AuthOk(
parseData(envelope.data, AuthOkPayload::class.java)
?: AuthOkPayload(userId = null, sessionId = null)
)
"chat_message" -> ServerMessage.ChatMessage(
parseData(envelope.data, ChatMessagePayload::class.java)
?: ChatMessagePayload(null, null, null, null)
)
"system_notice" -> ServerMessage.SystemNotice(
parseData(envelope.data, SystemNoticePayload::class.java)
?: SystemNoticePayload(title = null, content = null)
)
"pong" -> ServerMessage.Pong(
parseData(envelope.data, PongPayload::class.java)
)
"ack" -> ServerMessage.Ack(
AckPayload(
requestId = envelope.requestId,
code = envelope.code,
message = envelope.message
)
)
"error" -> ServerMessage.BusinessError(
code = envelope.code,
message = envelope.message,
requestId = envelope.requestId
)
else -> ServerMessage.Unknown(
type = type,
rawData = envelope.data
)
}
ParseResult.Success(message)
} catch (e: Throwable) {
ParseResult.Failure(raw, e)
}
}
private fun <T> parseData(data: JsonElement?, clazz: Class<T>): T? {
if (data == null || data.isJsonNull) return null
return gson.fromJson(data, clazz)
}
}
data class OutgoingEnvelope(
val type: String,
val requestId: String? = null,
val data: Any? = null,
val timestamp: Long = System.currentTimeMillis()
)
data class AuthPayload(val token: String)
data class ChatSendPayload(
val roomId: String,
val content: String
)
data class PingPayload(
val clientTime: Long = System.currentTimeMillis()
)
发送：
val json = gson.toJson(
OutgoingEnvelope(
type = "chat_send",
requestId = "req_${System.currentTimeMillis()}",
data = ChatSendPayload(roomId = "room_1", content = "hello")
)
)
webSocket.send(json)
不要把 newWebSocket 直接散落在 Activity 里。
封装层应负责：
enum class WsState {
IDLE,
CONNECTING,
CONNECTED,
AUTHENTICATING,
AUTHENTICATED,
CLOSING,
CLOSED,
FAILED
}
注意：
onOpen 只表示 WebSocket 连接成功。
如果服务端要求连接后发送 auth 消息，onOpen 不等于业务已登录。
class AndroidWebSocketClient(
private val url: String,
private val okHttpClient: OkHttpClient,
private val gson: Gson = Gson(),
private val mainHandler: Handler = Handler(Looper.getMainLooper())
) {
interface Listener {
fun onStateChanged(state: WsState) = Unit
fun onConnected() = Unit
fun onMessage(message: ServerMessage) = Unit
fun onRawMessage(text: String) = Unit
fun onParseError(raw: String, error: Throwable) = Unit
fun onAuthFailed(code: Int?, message: String?) = Unit
fun onError(error: Throwable, code: Int?, message: String?) = Unit
fun onDisconnected(code: Int, reason: String) = Unit
}
private val parser = WsMessageParser(gson)
@Volatile
private var webSocket: WebSocket? = null
@Volatile
private var state: WsState = WsState.IDLE
@Volatile
private var shouldReconnect: Boolean = true
private var retryCount: Int = 0
private var listener: Listener? = null
private var tokenProvider: (() -> String?)? = null
private val reconnectRunnable = Runnable {
connect()
}
private val internalListener = object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) {
this@AndroidWebSocketClient.webSocket = webSocket
retryCount = 0
changeState(WsState.CONNECTED)
postToMain {
listener?.onConnected()
}
}
override fun onMessage(webSocket: WebSocket, text: String) {
postToMain {
listener?.onRawMessage(text)
}
when (val result = parser.parse(text)) {
is WsMessageParser.ParseResult.Success -> {
postToMain {
listener?.onMessage(result.message)
}
}
is WsMessageParser.ParseResult.Failure -> {
postToMain {
listener?.onParseError(result.raw, result.error)
}
}
}
}
override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
this@AndroidWebSocketClient.webSocket = null
changeState(WsState.CLOSED)
postToMain {
listener?.onDisconnected(code, reason)
}
if (shouldReconnect) {
scheduleReconnect()
}
}
override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
this@AndroidWebSocketClient.webSocket = null
changeState(WsState.FAILED)
val code = response?.code
val message = response?.message
val authError = code == 401 || code == 403
if (authError) {
shouldReconnect = false
}
postToMain {
if (authError) {
listener?.onAuthFailed(code, message)
}
listener?.onError(t, code, message)
}
if (shouldReconnect) {
scheduleReconnect()
}
}
}
fun setListener(listener: Listener?) {
this.listener = listener
}
fun setTokenProvider(provider: (() -> String?)?) {
tokenProvider = provider
}
fun connect() {
if (state == WsState.CONNECTING || state == WsState.CONNECTED) return
shouldReconnect = true
mainHandler.removeCallbacks(reconnectRunnable)
changeState(WsState.CONNECTING)
val requestBuilder = Request.Builder().url(url)
val token = tokenProvider?.invoke()
if (!token.isNullOrBlank()) {
requestBuilder.addHeader("Authorization", "Bearer $token")
}
webSocket = okHttpClient.newWebSocket(
requestBuilder.build(),
internalListener
)
}
fun disconnect(code: Int = 1000, reason: String = "client close") {
shouldReconnect = false
mainHandler.removeCallbacks(reconnectRunnable)
changeState(WsState.CLOSING)
val closing = webSocket?.close(code, reason) ?: false
if (!closing) {
webSocket = null
changeState(WsState.CLOSED)
}
}
fun cancel() {
shouldReconnect = false
mainHandler.removeCallbacks(reconnectRunnable)
webSocket?.cancel()
webSocket = null
changeState(WsState.CLOSED)
}
fun send(type: String, data: Any? = null, requestId: String? = null): Boolean {
val current = webSocket ?: return false
if (state != WsState.CONNECTED && state != WsState.AUTHENTICATED) return false
val json = gson.toJson(
OutgoingEnvelope(
type = type,
requestId = requestId,
data = data
)
)
return current.send(json)
}
private fun scheduleReconnect() {
val delay = nextReconnectDelayMillis()
mainHandler.removeCallbacks(reconnectRunnable)
mainHandler.postDelayed(reconnectRunnable, delay)
}
private fun nextReconnectDelayMillis(): Long {
val cappedRetry = min(retryCount, 5)
val baseDelay = 1_000L * (1L shl cappedRetry)
val jitter = Random.nextLong(0L, 500L)
retryCount += 1
return min(baseDelay + jitter, 30_000L)
}
private fun changeState(newState: WsState) {
state = newState
postToMain {
listener?.onStateChanged(newState)
}
}
private fun postToMain(block: () -> Unit) {
if (Looper.myLooper() == Looper.getMainLooper()) {
block()
} else {
mainHandler.post(block)
}
}
}
这是学习型封装。
生产项目还要继续拆分心跳、重连、订阅、ACK、网络监听等职责。
不要在 onMessage 里直接更新 UI。
错误示例：
override fun onMessage(webSocket: WebSocket, text: String) {
textView.text = text
}
正确做法是切回主线程。
封装层可以统一处理：
private val mainHandler = Handler(Looper.getMainLooper())
private fun postToMain(block: () -> Unit) {
mainHandler.post(block)
}
Activity / Fragment
|
v
ViewModel
|
v
WebSocketRepository
|
v
AndroidWebSocketClient
职责：
| 层 | 职责 |
|---|---|
| Activity/Fragment | 展示 UI，收集状态 |
| ViewModel | 保存页面状态，调用 Repository |
| Repository | 处理连接、业务分发、HTTP 补偿 |
| WebSocketClient | 只处理底层连接和事件 |
| 场景 | 策略 |
|---|---|
| 只在单个页面实时 | 页面进入连接，离开断开 |
| 全 App 需要实时通知 | 放在 App 级 Repository 或 Service |
| 后台必须持续实时 | 考虑 ForegroundService 和系统限制 |
| 只是偶尔刷新 | 不一定需要 WebSocket |
不要无脑永久保持长连接。
要权衡：
class MainActivity : AppCompatActivity() {
private lateinit var wsClient: AndroidWebSocketClient
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
wsClient = AndroidWebSocketClient(
url = "wss://api.example.com/socket",
okHttpClient = webSocketOkHttpClient
)
wsClient.setTokenProvider {
authRepository.currentAccessToken()
}
wsClient.setListener(object : AndroidWebSocketClient.Listener {
override fun onStateChanged(state: WsState) {
renderConnectionState(state)
}
override fun onConnected() {
val token = authRepository.currentAccessToken()
if (!token.isNullOrBlank()) {
wsClient.send("auth", AuthPayload(token))
}
}
override fun onMessage(message: ServerMessage) {
when (message) {
is ServerMessage.AuthOk -> renderAuthOk(message.payload)
is ServerMessage.ChatMessage -> renderChatMessage(message.payload)
is ServerMessage.SystemNotice -> renderNotice(message.payload)
is ServerMessage.Pong -> Unit
is ServerMessage.Ack -> handleAck(message.payload)
is ServerMessage.BusinessError -> showError(message.message)
is ServerMessage.Unknown -> logUnknown(message.type)
}
}
override fun onAuthFailed(code: Int?, message: String?) {
refreshTokenAndReconnect()
}
})
}
override fun onStart() {
super.onStart()
wsClient.connect()
}
override fun onStop() {
super.onStop()
wsClient.disconnect()
}
}
长连接可能变成半开连接。
客户端本地对象还在，但真实网络链路已经坏了。
没有心跳时：
App 显示已连接。
实际连接已经被 NAT 清理。
服务端推送不到客户端。
客户端也不知道连接已经不可用。
心跳可以：
| 类型 | 所属层级 | App 是否容易控制 | 说明 |
|---|---|---|---|
| TCP keepalive | TCP | 不方便 | Android App 一般不直接依赖 |
| WebSocket ping/pong | WebSocket 协议 | 方便 | OkHttp pingInterval |
| 业务 JSON 心跳 | 业务协议 | 最方便 | {"type":"ping"} |
val client = OkHttpClient.Builder()
.pingInterval(20, TimeUnit.SECONDS)
.build()
流程：
OkHttp 每 20 秒发送 ping 控制帧。
服务端 WebSocket 协议栈回复 pong。
OkHttp 收到 pong，认为连接可用。
如果 pong 超时或写入失败，触发 onFailure。
注意：
客户端发送：
{
"type": "ping",
"requestId": "ping_1710000000000",
"data": {
"clientTime": 1710000000000
}
}
服务端返回：
{
"type": "pong",
"requestId": "ping_1710000000000",
"data": {
"serverTime": 1710000000100
}
}
业务心跳可以携带：
但心跳包应尽量小。
| 场景 | 参考间隔 |
|---|---|
| IM 前台聊天 | 15 到 30 秒 |
| 订单状态通知 | 30 到 60 秒 |
| 行情或游戏 | 5 到 15 秒，需要谨慎评估 |
| 后台弱实时 | 尽量降低频率，或改用推送 |
原则：
心跳间隔要小于服务端或网关 idle timeout。
如果网关 60 秒无数据就断开，可以设置 30 秒或 45 秒心跳。
private var lastPongAt: Long = 0L
private var missedPongCount: Int = 0
fun onPong() {
lastPongAt = System.currentTimeMillis()
missedPongCount = 0
}
fun checkHeartbeatTimeout() {
val now = System.currentTimeMillis()
val timeout = now - lastPongAt > 60_000L
if (timeout) {
missedPongCount += 1
}
if (missedPongCount >= 3) {
wsClient.cancel()
wsClient.connect()
}
}
不要一次心跳慢了就立即重连。
移动网络会抖动，连续失败多次再判断更稳。
class HeartbeatManager(
private val scope: CoroutineScope,
private val sendPing: () -> Boolean,
private val onTimeout: () -> Unit
) {
private var job: Job? = null
private var lastPongAt: Long = 0L
fun start() {
stop()
lastPongAt = System.currentTimeMillis()
job = scope.launch(Dispatchers.Default) {
while (isActive) {
delay(30_000L)
val sent = sendPing()
val timeout = System.currentTimeMillis() - lastPongAt > 60_000L
if (!sent || timeout) {
onTimeout()
}
}
}
}
fun onPong() {
lastPongAt = System.currentTimeMillis()
}
fun stop() {
job?.cancel()
job = null
}
}
启动和停止时机：
onOpen 后启动。onClosed 后停止。onFailure 后停止。不要失败后立即无限重连。
推荐指数退避：
1s -> 2s -> 4s -> 8s -> 16s -> 30s
加随机抖动 jitter，避免大量客户端同时重连。
伪代码：
onFailure/onClosed
-> 如果用户主动关闭，不重连
-> 如果 401/403，不盲目重连，先刷新 token
-> 如果网络不可用，等待网络恢复
-> 其他情况按指数退避重连
class NetworkMonitor(
private val context: Context,
private val onAvailable: () -> Unit,
private val onLost: () -> Unit
) {
private val connectivityManager =
context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
private val callback = object : ConnectivityManager.NetworkCallback() {
override fun onAvailable(network: Network) {
onAvailable()
}
override fun onLost(network: Network) {
onLost()
}
}
fun start() {
val request = NetworkRequest.Builder()
.addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
.build()
connectivityManager.registerNetworkCallback(request, callback)
}
fun stop() {
connectivityManager.unregisterNetworkCallback(callback)
}
}
网络断开时：
标记网络不可用。
可以 cancel 当前连接。
暂停重连。
网络恢复时：
检查登录态。
获取最新 token。
重新 connect。
| 分类 | 解决什么 |
|---|---|
| TLS/WSS | 通道是否安全，服务器是否可信 |
| token/cookie/session | 客户端是谁，有没有权限 |
wss:// 不代表已经登录。
token 不替代 TLS。
val request = Request.Builder()
.url("wss://api.example.com/socket")
.addHeader("Authorization", "Bearer $token")
.build()
流程：
登录拿 token。
WebSocket 握手 Header 带 token。
服务端校验 token。
通过返回 101。
失败返回 401/403 或拒绝升级。
token 过期：
onFailure response.code == 401/403
-> 停止自动重连
-> Retrofit 刷新 token
-> 刷新成功后重新 connect
-> 刷新失败跳登录
流程：
登录接口返回 Set-Cookie。
OkHttp CookieJar 保存 cookie。
WebSocket 握手携带 Cookie。
服务端根据 sessionId 查 session。
简单 CookieJar：
class InMemoryCookieJar : CookieJar {
private val cookieStore = mutableMapOf<String, List<Cookie>>()
override fun saveFromResponse(url: HttpUrl, cookies: List<Cookie>) {
cookieStore[url.host] = cookies
}
override fun loadForRequest(url: HttpUrl): List<Cookie> {
return cookieStore[url.host].orEmpty()
}
}
生产项目需要处理 domain、path、过期、持久化和退出登录清理。
有些业务允许先建立 WebSocket，再发送第一条鉴权消息。
{"type":"auth","data":{"token":"xxx"}}
这种方式属于业务协议。
注意：
wss:// 流程：
TCP 连接
-> TLS 握手
-> 校验证书链
-> 校验域名
-> HTTP Upgrade
-> WebSocket 帧通信
生产环境不要：
hostnameVerifier { _, _ -> true }
也不要使用信任所有证书的 X509TrustManager。
这些会破坏 TLS 安全性。
只有在自签证书、内网 CA、mTLS、证书锁定等场景才需要自定义 TLS 配置。
测试环境如果使用 ws://，可能需要：
<application
android:networkSecurityConfig="@xml/network_security_config">
</application>
res/xml/network_security_config.xml：
<?xml version="1.0" encoding="utf-8"?>
<network-security-config>
<domain-config cleartextTrafficPermitted="true">
<domain includeSubdomains="true">10.0.2.2</domain>
<domain includeSubdomains="true">example.test</domain>
</domain-config>
</network-security-config>
只建议调试环境使用。
WebSocket 是持续事件源。
事件包括：
Flow 适合表达持续异步事件流。
sealed class WsEvent {
data class StateChanged(val state: WsState) : WsEvent()
object Opened : WsEvent()
data class Message(val message: ServerMessage) : WsEvent()
data class RawMessage(val text: String) : WsEvent()
data class ParseError(val raw: String, val error: Throwable) : WsEvent()
data class Closed(val code: Int, val reason: String) : WsEvent()
data class Failure(val error: Throwable, val code: Int?, val message: String?) : WsEvent()
data class AuthFailed(val code: Int?, val message: String?) : WsEvent()
}
class FlowWebSocketClient(
private val okHttpClient: OkHttpClient,
private val parser: WsMessageParser
) {
fun connect(url: String, tokenProvider: () -> String?): Flow<WsEvent> = callbackFlow {
val requestBuilder = Request.Builder().url(url)
val token = tokenProvider()
if (!token.isNullOrBlank()) {
requestBuilder.addHeader("Authorization", "Bearer $token")
}
val listener = object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) {
trySend(WsEvent.StateChanged(WsState.CONNECTED))
trySend(WsEvent.Opened)
}
override fun onMessage(webSocket: WebSocket, text: String) {
trySend(WsEvent.RawMessage(text))
when (val result = parser.parse(text)) {
is WsMessageParser.ParseResult.Success -> {
trySend(WsEvent.Message(result.message))
}
is WsMessageParser.ParseResult.Failure -> {
trySend(WsEvent.ParseError(result.raw, result.error))
}
}
}
override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
trySend(WsEvent.StateChanged(WsState.CLOSED))
trySend(WsEvent.Closed(code, reason))
close()
}
override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
val code = response?.code
val message = response?.message
trySend(WsEvent.StateChanged(WsState.FAILED))
if (code == 401 || code == 403) {
trySend(WsEvent.AuthFailed(code, message))
}
trySend(WsEvent.Failure(t, code, message))
close(t)
}
}
trySend(WsEvent.StateChanged(WsState.CONNECTING))
val webSocket = okHttpClient.newWebSocket(requestBuilder.build(), listener)
awaitClose {
webSocket.close(1000, "flow closed")
}
}
}
class WsRepository(
private val flowClient: FlowWebSocketClient,
private val authRepository: AuthRepository
) {
private val _state = MutableStateFlow(WsState.IDLE)
val state: StateFlow<WsState> = _state
private val _messages = MutableSharedFlow<ServerMessage>(extraBufferCapacity = 64)
val messages: SharedFlow<ServerMessage> = _messages
private var collectJob: Job? = null
fun connect(scope: CoroutineScope) {
collectJob?.cancel()
collectJob = scope.launch {
flowClient.connect(
url = "wss://api.example.com/socket",
tokenProvider = { authRepository.currentAccessToken() }
).collect { event ->
when (event) {
is WsEvent.StateChanged -> _state.value = event.state
is WsEvent.Message -> _messages.emit(event.message)
is WsEvent.AuthFailed -> handleAuthFailed(scope)
is WsEvent.Failure -> handleFailure(event)
is WsEvent.Closed -> handleClosed(event)
else -> Unit
}
}
}
}
fun disconnect() {
collectJob?.cancel()
collectJob = null
_state.value = WsState.CLOSED
}
private fun handleAuthFailed(scope: CoroutineScope) {
scope.launch {
val refreshed = authRepository.refreshAccessToken()
if (refreshed) connect(scope)
}
}
private fun handleFailure(event: WsEvent.Failure) {
// 根据错误类型决定是否重连。
}
private fun handleClosed(event: WsEvent.Closed) {
// 根据 close code 决定是否重连。
}
}
lifecycleScope.launch {
repeatOnLifecycle(Lifecycle.State.STARTED) {
launch {
wsRepository.state.collect { state ->
renderState(state)
}
}
launch {
wsRepository.messages.collect { message ->
renderMessage(message)
}
}
}
}
repeatOnLifecycle 可以避免页面停止后继续更新 UI。
如果项目历史上大量使用 RxJava，可以继续用 RxJava 封装 WebSocket 事件。
如果是新 Kotlin 项目，通常优先协程 Flow。
| 概念 | 说明 |
|---|---|
Observable |
持续发射多个事件 |
Flowable |
支持背压的事件流 |
Single |
只返回一个结果或错误 |
Subject |
既是观察者也是被观察者 |
Disposable |
订阅句柄，页面销毁时清理 |
Scheduler |
线程调度 |
class RxWebSocketClient(
private val okHttpClient: OkHttpClient,
private val parser: WsMessageParser
) {
private val eventSubject = PublishSubject.create<WsEvent>().toSerialized()
private var webSocket: WebSocket? = null
fun events(): Observable<WsEvent> {
return eventSubject.hide()
}
fun connect(url: String, token: String?) {
val requestBuilder = Request.Builder().url(url)
if (!token.isNullOrBlank()) {
requestBuilder.addHeader("Authorization", "Bearer $token")
}
eventSubject.onNext(WsEvent.StateChanged(WsState.CONNECTING))
webSocket = okHttpClient.newWebSocket(
requestBuilder.build(),
object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) {
eventSubject.onNext(WsEvent.StateChanged(WsState.CONNECTED))
eventSubject.onNext(WsEvent.Opened)
}
override fun onMessage(webSocket: WebSocket, text: String) {
eventSubject.onNext(WsEvent.RawMessage(text))
when (val result = parser.parse(text)) {
is WsMessageParser.ParseResult.Success -> {
eventSubject.onNext(WsEvent.Message(result.message))
}
is WsMessageParser.ParseResult.Failure -> {
eventSubject.onNext(WsEvent.ParseError(result.raw, result.error))
}
}
}
override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
val code = response?.code
val message = response?.message
eventSubject.onNext(WsEvent.StateChanged(WsState.FAILED))
if (code == 401 || code == 403) {
eventSubject.onNext(WsEvent.AuthFailed(code, message))
}
eventSubject.onNext(WsEvent.Failure(t, code, message))
}
}
)
}
fun send(text: String): Boolean {
return webSocket?.send(text) ?: false
}
fun close() {
webSocket?.close(1000, "rx close")
webSocket = null
}
}
private val compositeDisposable = CompositeDisposable()
fun observeWebSocket() {
val disposable = rxWebSocketClient.events()
.observeOn(AndroidSchedulers.mainThread())
.subscribe { event ->
when (event) {
is WsEvent.StateChanged -> renderState(event.state)
is WsEvent.Message -> renderMessage(event.message)
is WsEvent.AuthFailed -> refreshTokenAndReconnect()
is WsEvent.Failure -> showError(event.error.message)
else -> Unit
}
}
compositeDisposable.add(disposable)
}
override fun onCleared() {
compositeDisposable.clear()
rxWebSocketClient.close()
}
interface RxApiService {
@POST("auth/login")
fun login(@Body request: LoginRequest): Single<LoginResponse>
@POST("auth/refresh")
fun refreshToken(@Body request: RefreshTokenRequest): Single<LoginResponse>
}
Retrofit：
val retrofit = Retrofit.Builder()
.baseUrl("https://api.example.com/")
.client(okHttpClient)
.addConverterFactory(GsonConverterFactory.create())
.addCallAdapterFactory(RxJava3CallAdapterFactory.create())
.build()
三种方式都能封装 WebSocket。
选择标准不是哪个更高级，而是哪个更适合当前项目。
| 方式 | 优点 | 缺点 | 适合场景 |
|---|---|---|---|
| 普通 callback | 简单直接，贴近 OkHttp 原始 API | 复杂后回调分散，状态组合麻烦 | Demo、小项目、底层封装内部 |
| 协程 Flow | Kotlin 官方生态，生命周期友好，和 ViewModel 配合自然 | 需要理解 Flow、SharedFlow、callbackFlow、Job | 新 Kotlin Android 项目 |
| RxJava | 操作符丰富，老项目生态成熟 | 学习成本高，要管理 Disposable 和背压 | 已大量使用 RxJava 的历史项目 |
推荐原则：
新 Kotlin 项目优先协程 Flow。
老项目如果已经全面使用 RxJava，就继续 RxJava。
底层可以仍然使用 OkHttp callback，但 Repository 对外只暴露一种模型。
不要让同一条 WebSocket 链路同时对外暴露 callback、Flow、Observable 三套接口。
这样会让状态同步、生命周期和错误处理变复杂。
WebSocket 是全双工，响应可能乱序回来。
所以发送需要带 requestId。
{"type":"subscribe","requestId":"req_10001","data":{"topic":"room_1"}}
服务端 ACK：
{"type":"ack","requestId":"req_10001","code":0,"message":"ok"}
class PendingRequestManager {
private val pending = mutableMapOf<String, CompletableDeferred<AckPayload>>()
fun create(requestId: String): CompletableDeferred<AckPayload> {
val deferred = CompletableDeferred<AckPayload>()
pending[requestId] = deferred
return deferred
}
fun complete(ack: AckPayload) {
val requestId = ack.requestId ?: return
pending.remove(requestId)?.complete(ack)
}
fun failAll(error: Throwable) {
pending.values.forEach { it.completeExceptionally(error) }
pending.clear()
}
}
发送并等待：
suspend fun sendAndAwaitAck(
type: String,
data: Any?,
timeoutMillis: Long = 5_000L
): AckPayload {
val requestId = "req_${System.currentTimeMillis()}"
val deferred = pendingRequestManager.create(requestId)
val sent = wsClient.send(type, data, requestId)
if (!sent) throw IllegalStateException("send failed")
return withTimeout(timeoutMillis) {
deferred.await()
}
}
很多服务端订阅关系绑定在当前连接上。
连接断开后，服务端会清理订阅。
重连后要重新订阅。
class SubscriptionManager {
private val topics = linkedSetOf<String>()
fun add(topic: String) {
topics.add(topic)
}
fun remove(topic: String) {
topics.remove(topic)
}
fun all(): List<String> {
return topics.toList()
}
}
恢复订阅：
fun onWebSocketConnected() {
subscriptionManager.all().forEach { topic ->
wsClient.send(
type = "subscribe",
data = mapOf("topic" to topic),
requestId = "sub_${System.currentTimeMillis()}"
)
}
}
WebSocket 断线期间，服务端推送的消息客户端收不到。
常见方案：
客户端保存 lastMessageId。
重连成功后用 Retrofit 调 HTTP 接口。
服务端返回 lastMessageId 之后的消息。
客户端按 messageId 去重后插入本地列表。
接口：
interface MessageApi {
@GET("chat/messagesAfter")
suspend fun getMessagesAfter(
@Query("roomId") roomId: String,
@Query("lastMessageId") lastMessageId: String
): List<ChatMessagePayload>
}
幂等表示同一操作执行一次和多次，最终结果一致。
WebSocket 重试中很重要。
建议：
messageId。messageId 去重。enum class SendState {
SENDING,
SENT,
FAILED
}
流程：
用户点击发送。
本地插入消息，状态 SENDING。
webSocket.send 成功入队。
等待服务端 ACK。
ACK code == 0，状态 SENT。
超时或 code != 0，状态 FAILED。
webSocket.queueSize() 表示待发送队列大小。
如果持续增长，说明发送速度大于网络实际发送能力。
处理方式：
背压表示：
生产速度 > 消费速度。
高频行情、设备状态、日志流都可能遇到。
处理方式：
buffer 或 conflate。Flow 示例：
messages
.conflate()
.collect { latest ->
renderLatestState(latest)
}
防抖 debounce：一段时间多次触发，只处理最后一次。
节流 throttle：一段时间最多处理一次。
WebSocket 高频 UI 刷新更常用节流。
例如服务端每秒 100 条行情，UI 每秒刷新 10 次可能就够了。
Node.js echo server：
const WebSocket = require('ws')
const server = new WebSocket.Server({ port: 8080 })
server.on('connection', (socket) => {
socket.send(JSON.stringify({
type: 'system_notice',
data: { title: 'connected', content: 'welcome' },
timestamp: Date.now()
}))
socket.on('message', (message) => {
socket.send(JSON.stringify({
type: 'chat_message',
data: {
roomId: 'room_1',
senderId: 'server',
content: message.toString(),
messageId: String(Date.now())
},
timestamp: Date.now()
}))
})
})
console.log('ws://localhost:8080')
Android 模拟器访问电脑本机：
ws://10.0.2.2:8080
| 现象 | 可能原因 | 排查方向 |
|---|---|---|
onOpen 不回调 |
URL 错误、网络不通、服务端拒绝 | 看 onFailure 和服务端日志 |
401 |
token 无效或过期 | 刷新 token，检查 Header |
403 |
无权限 | 检查账号权限和业务配置 |
404 |
WebSocket 路径错误 | 检查 URL、网关配置 |
| TLS handshake failed | 证书或域名问题 | 检查证书链、域名、系统时间 |
| cleartext not permitted | Android 禁止明文 | 用 wss 或配置 network security config |
| JSON 解析失败 | 协议不一致 | 打印 raw，检查 type/data |
| send 返回 false | 未连接或正在关闭 | 等 onOpen 后发送 |
| 频繁重连 | 心跳失败、服务端断开、网络不稳 | 看 close code、reason、心跳间隔 |
| 后台断开 | 系统限制或 NAT 超时 | 前后台策略、ForegroundService、心跳 |
| UI 崩溃 | 非主线程更新 UI | 切主线程 |
Gson 依赖反射，混淆时要注意 DTO。
建议：
@SerializedName。-keepattributes Signature
-keepattributes *Annotation*
-keep class com.example.app.ws.model.** { *; }
应该记录：
不应该记录：
token 脱敏：
fun maskToken(token: String): String {
if (token.length <= 8) return "****"
return token.take(4) + "****" + token.takeLast(4)
}
连接层：
wss://。鉴权层：
消息层：
生命周期层：
安全层：
测试层：
com.example.app.websocket
├── AndroidWebSocketClient.kt
├── WsState.kt
├── WsModels.kt
├── WsMessageParser.kt
├── WsRepository.kt
├── HeartbeatManager.kt
├── ReconnectPolicy.kt
├── SubscriptionManager.kt
├── PendingRequestManager.kt
├── NetworkMonitor.kt
└── WebSocketConfig.kt
职责：
| 文件 | 职责 |
|---|---|
AndroidWebSocketClient.kt |
底层连接、发送、关闭 |
WsModels.kt |
Envelope、payload、ServerMessage |
WsMessageParser.kt |
Gson 解析 |
WsRepository.kt |
业务连接管理 |
HeartbeatManager.kt |
业务心跳 |
ReconnectPolicy.kt |
重连延迟和策略 |
SubscriptionManager.kt |
订阅保存和恢复 |
PendingRequestManager.kt |
requestId 和 ACK 匹配 |
NetworkMonitor.kt |
网络变化监听 |
WebSocketConfig.kt |
URL、心跳、超时、重连参数 |
HTTP 最初是无状态请求响应。
登录和购物车需要状态，于是出现 session/cookie。
服务器保存 session 有扩展压力，于是 token/JWT 常用于无状态鉴权。
实时业务需要服务端主动推送，于是 WebSocket 替代轮询。
WebSocket 先 HTTP Upgrade，再进入双向帧通信。
wss:// 负责通道安全，token/cookie/session 负责身份权限。
Android 中 Retrofit 负责 HTTP API，OkHttp newWebSocket 负责长连接。
Gson 负责 JSON 和 Kotlin 对象转换。
协程 Flow 或 RxJava 可以把 WebSocket 回调组织成事件流。
生产可用的 WebSocket 必须处理心跳、重连、ACK、补偿、去重、生命周期和安全。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。