Android Kotlin中Socket.IO的Ack回调在Future函数中未触发

问题场景 #

开发Android Kotlin项目时，使用Socket.IO与Node.js后端通信，封装了request函数用于发送事件并通过Ack回调获取结果，但函数内的Ack回调从未触发；直接调用socket.emit传入Ack回调却能正常工作。核心需求是不使用协程，通过Ack回调获取结果并赋值给变量。

问题代码 #

Kotlin客户端代码：

lateinit var socket: Socket

fun startMic() {
    val opts = IO.Options()
    opts.path = "/mic-io"
    opts.query = ""
    opts.forceNew = true
    opts.reconnection = true
    socket = IO.socket("http://192.168.1.3:9600", opts)
    if (!socket.connected()) {
        Log.i("MICTAG", "CONNECTING")
        socket.connect()
    }

    socket.off(Socket.EVENT_CONNECT).on(Socket.EVENT_CONNECT) {
        Log.i("MICTAG", "CONNECTED")
        val res = request("getRtpCapabilities")
        Log.i("MICTAG", "DONE")
    }
}

fun request(event: String): Array<Any?>? {
    val future = CompletableFuture<Array<Any?>?>()
    Log.i("MICTAG", "SENDING")
    socket.emit(event, Ack { data: Array<Any?>? ->
        // 永远不会执行到这里
        Log.i("MICTAG","new data using future")
        future.complete(data)
    })

    return future.join()
}

日志输出：

- CONNECTING
- CONNECTED
- SENDING

直接调用能正常触发回调的代码：

socket.emit("getRtpCapabilities", Ack { data: Array<Any?>? ->
    msg("new data using callback")
})

Node.js服务器代码：

socket.on('getRtpCapabilities', (callback) => {
    console.log('new data');
    callback({ ok: 'success' });
});

问题原因 #

request函数中调用future.join()会阻塞当前线程，而Socket.IO的Ack回调依赖同一个线程（Socket.IO事件循环线程）调度执行。当EVENT_CONNECT回调执行时，调用request后线程被join()阻塞，Ack回调无法被调度，形成死锁，因此永远无法进入Ack代码块。

直接调用socket.emit时没有阻塞线程，回调能被事件循环正常调度，所以可以正常触发。

解决方案 #

方案1：回调式设计（推荐） #

避免阻塞线程，将request函数改为回调式，通过参数传递结果：

lateinit var socket: Socket

fun startMic() {
    val opts = IO.Options()
    opts.path = "/mic-io"
    opts.query = ""
    opts.forceNew = true
    opts.reconnection = true
    socket = IO.socket("http://192.168.1.3:9600", opts)
    if (!socket.connected()) {
        Log.i("MICTAG", "CONNECTING")
        socket.connect()
    }

    socket.off(Socket.EVENT_CONNECT).on(Socket.EVENT_CONNECT) {
        Log.i("MICTAG", "CONNECTED")
        // 通过回调获取结果
        request("getRtpCapabilities") { data ->
            Log.i("MICTAG", "DONE, data: ${data?.contentToString()}")
            // 在这里处理返回数据，比如赋值给变量
            val res = data
        }
    }
}

// 调整为带回调参数的request函数
fun request(event: String, onResult: (Array<Any?>?) -> Unit) {
    Log.i("MICTAG", "SENDING")
    socket.emit(event, Ack { data: Array<Any?>? ->
        Log.i("MICTAG","new data using callback")
        onResult(data)
    })
}

方案2：单独线程执行（不推荐） #

如果一定要同步获取结果，可将emit操作放到单独线程，避免阻塞Socket.IO事件循环：

fun request(event: String): Array<Any?>? {
    val future = CompletableFuture<Array<Any?>?>()
    // 在独立线程中执行emit
    Thread {
        socket.emit(event, Ack { data: Array<Any?>? ->
            Log.i("MICTAG","new data using future")
            future.complete(data)
        })
    }.start()
    return future.join()
}

注意：方案2可能带来线程切换的额外问题（如UI操作需切回主线程），优先推荐回调式设计。

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