Android中基于callbackFlow的WebSocket重连及后台保活问题

重构WebSocketRepository：统一管理连接与重连 #

原实现中每次收集socketEventsFlow()都会创建新连接，且Flow生命周期绑定到收集者，导致后台时连接被关闭。我们将WebSocket生命周期与收集者解耦，用SharedFlow分发事件并内置重连逻辑：

class WebSocketRepository {
    // 用SharedFlow分发事件，支持多订阅者
    private val _events = MutableSharedFlow<GameServerResponse>(replay = 0)
    val events: SharedFlow<GameServerResponse> = _events

    // 独立协程作用域管理连接，不受收集者生命周期影响
    private val socketScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
    private var socket: WebSocket? = null
    private var isReconnecting = false
    private var reconnectDelay = 1000L
    private val maxReconnectDelay = 30000L

    init {
        // 初始化时启动连接
        connectWebSocket()
    }

    private fun connectWebSocket() {
        if (isReconnecting) return
        isReconnecting = true

        val socketListener = object : WebSocketListener() {
            override fun onOpen(webSocket: WebSocket, response: Response) {
                reconnectDelay = 1000L
                isReconnecting = false
                socket = webSocket
            }

            override fun onMessage(webSocket: WebSocket, text: String) {
                val message = Gson().fromJson(text, GameServerResponse::class.java)
                socketScope.launch {
                    _events.emit(message)
                }
            }

            override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
                handleDisconnection()
            }

            override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
                // 非主动关闭时触发重连
                if (code != 1000) {
                    handleDisconnection()
                }
                socket = null
            }
        }

        val client = OkHttpClient()
        val request = Request.Builder()
            .url("ws://10.0.2.2:8080")
            .build()

        socket = client.newWebSocket(request, socketListener)
    }

    private fun handleDisconnection() {
        if (isReconnecting) return
        socketScope.launch {
            // 指数退避延迟，避免频繁重试
            delay(reconnectDelay)
            reconnectDelay = minOf(reconnectDelay * 2, maxReconnectDelay)
            connectWebSocket()
        }
    }

    // 应用完全退出时主动关闭连接
    fun closeSocket() {
        socket?.close(1000, "Application terminated")
        socket = null
        socketScope.cancel()
    }
}

调整ViewModel：持有稳定的事件流 #

ViewModel只需暴露Repository的事件流，无需每次创建新Flow：

class MainActivityViewModel @Inject constructor(
    private val repository: WebSocketRepository
) : ViewModel() {
    val events = repository.events

    // ViewModel销毁时主动关闭Socket
    override fun onCleared() {
        super.onCleared()
        repository.closeSocket()
    }
}

修改Activity收集逻辑：适配后台保活 #

现在Socket生命周期由Repository独立管理，Activity只需安全收集事件，后台时仅暂停收集，不会关闭连接：

@AndroidEntryPoint
class MainActivity : AppCompatActivity() {
    private lateinit var binding: ActivityMainBinding
    private val viewModel: MainActivityViewModel by viewModels()

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        binding = ActivityMainBinding.inflate(layoutInflater)
        setContentView(binding.root)

        lifecycleScope.launch {
            lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
                viewModel.events.collect { event ->
                    // 处理事件，如启动Fragment等
                }
            }
        }
    }
}

问题解决说明 #

1. 后台不关闭Socket #

• 原callbackFlow在收集取消时（后台触发repeatOnLifecycle(STARTED)暂停）会执行awaitClose关闭Socket，现在将Socket生命周期与收集者解耦，由Repository的独立协程作用域管理，仅在应用完全退出（ViewModel销毁）时主动关闭。
• 后台时Activity仅暂停收集事件，Socket保持连接，回到前台后自动恢复收集，无需重建连接。

2. 自动重连实现 #

• 在onFailure和非主动关闭的onClosed回调中触发重连，采用指数退避策略（延迟翻倍至最大值），避免频繁重试消耗资源。
• 用isReconnecting标记防止重复触发重连任务，连接成功后重置延迟，恢复正常重连节奏。

内容的提问来源于stack exchange，提问作者lela2011