WebSocket实战：OkHttp WebSocket客户端、使用Flow封装WebSocket消息、心跳机制

聊到现代Android网络层，HTTP请求只是基本功。真正让应用“活”起来的，是实时通信能力。WebSocket就是那个让服务器能主动找你的技术。

我最早接触WebSocket是在做一个即时通讯模块时。当时用轮询，每3秒发一次请求，用户量一上来，服务器直接扛不住。换成WebSocket后，连接数从每秒几千次请求降到了几十个长连接，效果立竿见影。

今天我们就来实战一下：用OkHttp实现WebSocket客户端，再用Kotlin Flow把消息封装成响应式流，最后加上心跳机制保活。

WebSocket vs HTTP：到底差在哪？

说白了，HTTP是“你问我答”的模式。客户端发请求，服务器给响应，完事。如果你想拿新数据，就得再问一次。

WebSocket不一样。它先通过HTTP握手建立连接，然后双方都可以随时发消息。服务器有新数据了，直接推给你。你想想看，这对聊天、股票行情、游戏同步这类场景有多重要。

我习惯把HTTP比作打电话问路，WebSocket比作对讲机。前者每次都要拨号，后者挂上频道就能随时说话。

OkHttp WebSocket客户端：从握手到收发

OkHttp本身就内置了WebSocket支持，不需要额外引入库。我们直接看代码。

class WebSocketClient(
    private val url: String,
    private val client: OkHttpClient = OkHttpClient.Builder()
        .readTimeout(0, TimeUnit.MILLISECONDS)  // WebSocket需要长连接，超时设为0
        .build()
) {
    private var webSocket: WebSocket? = null

    fun connect(listener: WebSocketListener) {
        val request = Request.Builder()
            .url(url)
            .build()
        webSocket = client.newWebSocket(request, listener)
    }

    fun send(message: String): Boolean {
        return webSocket?.send(message) ?: false
    }

    fun close(code: Int = 1000, reason: String = "normal close") {
        webSocket?.close(code, reason)
    }
}

这里有个坑，我一开始就踩过。OkHttp的readTimeout默认是10秒。WebSocket是长连接，10秒没数据就断开了。所以必须设为0，表示不超时。

WebSocketListener是核心回调接口，我们需要实现这几个方法：

val listener = object : WebSocketListener() {
    override fun onOpen(webSocket: WebSocket, response: Response) {
        println("连接已建立")
    }

    override fun onMessage(webSocket: WebSocket, text: String) {
        println("收到消息: $text")
    }

    override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
        println("连接失败: ${t.message}")
    }

    override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
        webSocket.close(code, reason)
    }

    override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
        println("连接已关闭")
    }
}

嗯，这里要注意：onClosing和onClosed是两个不同的阶段。前者是对方发起关闭请求，你还有机会做清理；后者是连接已经彻底断了。

使用Flow封装WebSocket消息

回调的方式用起来不够爽。在Kotlin的世界里，Flow才是王道。我们可以把WebSocket的消息流封装成Flow，让调用方用响应式的方式消费数据。

我封装的时候，主要考虑三个点：连接状态、收到的消息、错误信息。把它们都塞进一个密封类里：

sealed class WebSocketEvent {
    data class Connected(val response: Response) : WebSocketEvent()
    data class Message(val text: String) : WebSocketEvent()
    data class Error(val throwable: Throwable) : WebSocketEvent()
    object Closed : WebSocketEvent()
}

然后创建一个Flow版本的WebSocket客户端：

class FlowWebSocketClient(private val url: String) {

    private val client = OkHttpClient.Builder()
        .readTimeout(0, TimeUnit.MILLISECONDS)
        .build()

    fun connect(): Flow<WebSocketEvent> = callbackFlow {
        val request = Request.Builder().url(url).build()

        val listener = object : WebSocketListener() {
            override fun onOpen(webSocket: WebSocket, response: Response) {
                trySend(WebSocketEvent.Connected(response))
            }

            override fun onMessage(webSocket: WebSocket, text: String) {
                trySend(WebSocketEvent.Message(text))
            }

            override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
                trySend(WebSocketEvent.Error(t))
            }

            override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
                trySend(WebSocketEvent.Closed)
                channel.close()
            }
        }

        val webSocket = client.newWebSocket(request, listener)

        awaitClose {
            webSocket.close(1000, "flow closed")
        }
    }
}

这里用了callbackFlow，它是专门用来把回调式API转成Flow的。注意awaitClose块，当Flow被取消时，这里会执行清理工作。

调用方用起来就清爽多了：

val client = FlowWebSocketClient("wss://example.com/ws")

lifecycleScope.launch {
    client.connect().collect { event ->
        when (event) {
            is WebSocketEvent.Connected -> println("连上了")
            is WebSocketEvent.Message -> handleMessage(event.text)
            is WebSocketEvent.Error -> handleError(event.throwable)
            is WebSocketEvent.Closed -> println("断开了")
        }
    }
}

你看，用Flow之后，整个数据流变得非常清晰。而且天然支持协程取消，不会泄漏。

心跳机制：让连接活得更久

WebSocket连接不是永久有效的。网络波动、NAT超时、运营商策略，都可能导致连接被静默断开。你这边还觉得连着，实际上服务器早就不认你了。

我遇到过最坑的一次：用户反馈消息发不出去，但App显示连接正常。查了半天，原来是移动4G网络下，NAT表项5分钟就过期了。从那以后，心跳机制就成了我WebSocket代码里的标配。

心跳的原理很简单：客户端定时发一个ping帧，服务器回复pong帧。如果一段时间没收到pong，就认为连接断了，触发重连。

OkHttp的WebSocket支持原生ping/pong：

class HeartbeatWebSocketClient(
    private val url: String,
    private val heartbeatInterval: Long = 30_000L,  // 30秒一次心跳
    private val heartbeatTimeout: Long = 10_000L    // 10秒没收到pong就算超时
) {
    private var webSocket: WebSocket? = null
    private var heartbeatJob: Job? = null
    private var lastPongTime: Long = 0L

    fun connect(scope: CoroutineScope) {
        val request = Request.Builder().url(url).build()
        val listener = object : WebSocketListener() {
            override fun onOpen(webSocket: WebSocket, response: Response) {
                lastPongTime = System.currentTimeMillis()
                startHeartbeat(scope, webSocket)
            }

            override fun onMessage(webSocket: WebSocket, text: String) {
                // 处理业务消息
            }

            override fun onPong(webSocket: WebSocket, payload: ByteString) {
                lastPongTime = System.currentTimeMillis()
            }

            override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
                stopHeartbeat()
                // 触发重连逻辑
            }
        }

        webSocket = OkHttpClient.Builder()
            .readTimeout(0, TimeUnit.MILLISECONDS)
            .build()
            .newWebSocket(request, listener)
    }

    private fun startHeartbeat(scope: CoroutineScope, ws: WebSocket) {
        heartbeatJob = scope.launch {
            while (isActive) {
                delay(heartbeatInterval)

                val elapsed = System.currentTimeMillis() - lastPongTime
                if (elapsed > heartbeatTimeout) {
                    // 超时了，关闭当前连接
                    ws.close(1001, "heartbeat timeout")
                    stopHeartbeat()
                    // 触发重连
                    break
                }

                // 发送ping帧
                ws.send(ByteString.encodeUtf8("ping"))
            }
        }
    }

    private fun stopHeartbeat() {
        heartbeatJob?.cancel()
        heartbeatJob = null
    }
}

这里有个细节：onPong回调是OkHttp自动处理的，你不需要手动回复pong。我们只需要记录收到pong的时间，用来判断是否超时。

心跳间隔怎么设？我一般建议30秒到60秒。太频繁浪费流量，太慢又可能被运营商掐断。你可以根据实际网络环境调整。

整体架构图

下面这张图展示了WebSocket客户端的完整工作流程：

WebSocket客户端架构与心跳机制 应用层 Flow封装层 OkHttp WebSocket TCP/网络层 心跳模块 重连模块 心跳流程 1. 每30秒发送ping 2. 记录pong到达时间 3. 超时10秒触发重连 4. 重连后重置心跳 重连策略 • 指数退避：1s, 2s, 4s... • 最大间隔：30秒 • 重连次数限制：10次 Flow封装层将回调转为响应式流，心跳模块独立运行，与业务层解耦

避坑指南

我总结几个实战中容易踩的坑：

  • 不要在主线程创建WebSocket。OkHttp的WebSocket内部会开线程，但回调默认在子线程。如果你要更新UI，记得切到主线程。
  • 重连要有退避策略。我曾经直接写死1秒重连，结果网络一波动，客户端疯狂重连，服务器直接被打挂了。用指数退避，第一次等1秒，第二次2秒，第三次4秒，最大不超过30秒。
  • Flow的取消要处理好。如果用callbackFlow，一定要在awaitClose里关闭WebSocket。否则协程取消了，连接还挂着，内存泄漏。
  • 心跳不要和业务消息混在一起。我见过有人用业务消息当心跳，结果业务一停，心跳也停了。心跳应该是独立的ping/pong帧，和业务完全解耦。
小技巧：如果你需要调试WebSocket，可以用Chrome的开发者工具。在Network标签页里，WebSocket的帧都能看到。我排查问题时经常用这个。
注意：WebSocket的URL是ws://或wss://开头。生产环境一定要用wss://，否则数据明文传输，安全风险很大。

好了，WebSocket实战这块就聊到这。核心就是三件事：用OkHttp建立连接、用Flow封装消息流、用心跳保活。你想想看，这三件事做好了，一个稳定可靠的实时通信层就有了。

我个人习惯把WebSocket封装成一个独立的模块，和网络请求库分开。这样职责清晰，测试也方便。你在项目里也可以试试这种思路。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →