Kotlin中Socket服务器断开与ConnectionResetException异常处理问题

核心问题 #

重连成功后，原来的Receiver线程仍持有旧连接的DataInputStream，当循环继续执行read()操作时，旧连接已被服务器重置，因此抛出ConnectionResetException。本质是旧线程未被正确终止，还在尝试读取失效的流。

解决步骤 #

1. 明确捕获ConnectionResetException

ConnectionResetException是SocketException的子类，可单独捕获或在SocketException分支内统一处理，避免遗漏异常类型：

catch (e: ConnectionResetException) {
    handleDisconnection()
} catch (e: SocketException) {
    handleDisconnection()
}

2. 重连时强制终止旧Receiver线程

在reconnect()方法中，先停止旧的Receiver线程，确保它不再继续运行：

private fun reconnect() {
    // 先终止旧Receiver线程
    receiver?.apply {
        running = false
        try {
            dataInputStream.close() // 关闭旧输入流，触发read()抛出异常退出循环
            join() // 等待线程完全退出
        } catch (e: Exception) {
            Log.d("Reconnect", "Failed to stop old receiver: ${e.message}")
        }
    }

    // 关闭旧连接
    try {
        client?.close()
        client = null
        sender = null
        receiver = null

        // 创建新连接
        client = Socket()
        val address = InetSocketAddress("192.168.***.***", 8888)
        client!!.connect(address, 20000)

        // 初始化新的Sender和Receiver
        val outputStream = client?.getOutputStream()
        val inputStream = client?.getInputStream()
        sender = Sender(DataOutputStream(outputStream))
        receiver = Receiver(DataInputStream(inputStream))

        receiver?.dataReceivedListeners?.add { data ->
            dataReceivedListeners.forEach { listener ->
                listener(data)
            }
        }
        receiver?.start()

        reconnected = true
        Log.d("Reconnection", "Reconnection succeeded")
    } catch (e: Exception) {
        reconnected = false
        Log.d("Reconnection failed", "Attempt failed: ${e.message}")
    }
}

3. 修正Receiver线程的退出逻辑

给running变量加上volatile修饰，保证多线程下的状态可见性，同时优化断开后的重连逻辑：

private inner class Receiver(private val dataInputStream: DataInputStream) : Thread() {
    val dataReceivedListeners = mutableListOf<(ByteArray) -> Unit>()
    @Volatile private var running = true // 线程安全的状态标记

    override fun run() {
        val buffer = ByteArray(1024)
        var bytesRead: Int

        while (running) {
            try {
                bytesRead = dataInputStream.read(buffer)
                if (bytesRead <= 0) {
                    break
                }
                // 数据处理逻辑...
            } catch (e: ConnectionResetException) {
                handleDisconnection()
            } catch (e: SocketException) {
                handleDisconnection()
            } catch (e: IOException) {
                Log.d("Receiver", "IO error: ${e.message}")
                running = false
            }
        }
    }

    private fun handleDisconnection() {
        if (reconnected) {
            reconnected = false
            var tries = 0
            while (tries < 5 && !reconnected) {
                reconnect()
                tries++
                Thread.sleep(1000)
            }
            if (tries >= 5) {
                Log.d("Reconnection", "All attempts failed")
                running = false // 彻底终止线程
            }
        }
    }
}

4. 线程安全管理reconnected变量

同样给reconnected加上volatile修饰，确保多线程下的状态同步：

@Volatile var reconnected = true

额外建议 #

• 避免在Receiver线程内直接执行重连逻辑，可通过回调通知主线程处理，减少线程嵌套风险。
• 重连时务必确保所有旧资源（Socket、流、线程）都被正确关闭，防止资源泄漏。

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