Kotlin如何实现带异步读取及读完回调的异步TCP Socket连接

问题描述 #

目前我的安卓应用通过TCP Socket与服务端通信，最初采用的是同步读写方法，代码如下：

private val addressString : String = "192.168.1.200"
private val port : Int = 33322
private var socketAddress: InetSocketAddress = InetSocketAddress(addressString, port)
private var socket: Socket? = null

fun connect(){
    socket = Socket()
    socketAddress = InetSocketAddress(addressString, port)
    try{
            socket!!.connect(socketAddress)
    }catch(e : Exception){
          socket!!.close()
    }
}

fun disconnect(){
     try {
        socket?.shutdownOutput()
    } catch (e: Exception) {}
    try {
        socket?.shutdownInput()
    } catch (e: Exception) {}
    try {
        socket?.close()
    } catch (e: Exception) {}
    socket = null
}

 fun send(data: ByteArray): Int {
    var sentByteCount: Int = -1
    try {
        socket!!.getOutputStream().write(data)
        sentByteCount = data.size
    } catch (e: Exception) {
        throw e
    }
    return sentByteCount
}

data class Wrapper<T>(
var value:T
)

  fun receive(buffer: Wrapper<ByteArray>): Int {
    val size = buffer.value.size
    var receivedByteCount: Int = -1
    try {
        receivedByteCount = socket!!.getInputStream().read(buffer.value)
    } catch (e: Exception) {
        throw e
    }
    return receivedByteCount
}

服务端由C#编写，采用异步读取模式和其他设备通信，对应代码如下：

public const int BUFFER_SIZE = 4096;
private string addressString = "192.168.1.200";
private int port = 33322;
private int timeout = 5000;
private byte[] _buffer = new byte[BUFFER_SIZE];
private TcpClient _client;
private SocketError _socketError;
private AsyncCallback _callback = new AsyncCallback(ReceiveCallback);

public void connect()
{
  _client = new TcpClient();
  _client.ConnectAsync(addressString, port).Wait(timeout);
}

public void receive()
{
    _client.BeginReceive(_buffer, 0, _buffer.Length, SocketFlags.None, out _socketError, _callback, null);
}

private void ReceiveCallback(IAsyncResult ar)
{
    // 异步读取完成后的回调逻辑
}

public void send()
{
     _client.GetStream().Write(_buffer, 0, _buffer.Length);
}

public void disconnect()
{
    if (_client.Connected)
     _client.Close();
}

我需要让我的安卓应用适配服务端现有的通信模式。之前查资料看到Kotlin可以用AsynchronousSocketChannel实现异步连接，但文档说这个Socket要求服务端必须是AsynchronousServerSocketChannel，我只能开发客户端，无法修改服务端实现。请问Kotlin有没有其他方案，可以实现和上述C#代码类似的异步Socket通信能力？

解决方案 #

首先纠正一个认知误区：TCP是传输层标准协议，两端的同步/异步实现逻辑互不影响，你之前看到的文档描述是误解，AsynchronousSocketChannel作为客户端异步API，完全可以对接任意标准TCP服务端，包括你现在用C#写的异步服务端，不需要服务端做任何修改。

除此之外，还有两种更适合安卓Kotlin开发场景的实现方案：

方案1：Kotlin协程封装现有同步代码（改造成本最低） #

你现有的同步Socket代码不需要做核心逻辑修改，只需要用Kotlin协程的IO调度器封装成异步调用，就能实现和C#异步回调/await完全一致的效果，示例代码如下：

import kotlinx.coroutines.*
import java.net.Socket

class TcpClient {
    private val addressString : String = "192.168.1.200"
    private val port : Int = 33322
    private val socketAddress = InetSocketAddress(addressString, port)
    private var socket: Socket? = null
    private val ioScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
    private val buffer = ByteArray(4096) // 和C#服务端缓冲区大小对齐

    // 异步连接，支持回调或await
    suspend fun connect(): Result<Unit> = withContext(Dispatchers.IO) {
        runCatching {
            socket = Socket()
            socket!!.connect(socketAddress, 5000) // 5秒超时和C#对齐
            startReceiveLoop() // 连接成功后启动异步接收循环
        }
    }

    // 异步发送
    suspend fun send(data: ByteArray): Result<Int> = withContext(Dispatchers.IO) {
        runCatching {
            socket?.getOutputStream()?.write(data) ?: throw IllegalStateException("未连接")
            data.size
        }
    }

    // 异步接收循环，收到数据后触发回调，和C#的ReceiveCallback逻辑对齐
    private fun startReceiveLoop() {
        ioScope.launch {
            while (isActive && socket?.isConnected == true) {
                runCatching {
                    val readCount = socket!!.getInputStream().read(buffer)
                    if (readCount <= 0) {
                        // 连接断开，处理断线逻辑
                        disconnect()
                        return@launch
                    }
                    // 处理收到的buffer，和C#回调里的逻辑一致
                    onReceiveData(buffer.copyOf(readCount))
                }.onFailure {
                    // 处理读取异常
                    disconnect()
                }
            }
        }
    }

    // 接收数据回调，替换为你自己的业务逻辑
    private fun onReceiveData(data: ByteArray) {
        // 处理收到的字节数据
    }

    fun disconnect() {
        ioScope.cancel()
        runCatching { socket?.shutdownOutput() }
        runCatching { socket?.shutdownInput() }
        runCatching { socket?.close() }
        socket = null
    }
}

使用的时候可以直接用协程await，也可以自己加回调包装，完全不会阻塞主线程，和C#的异步调用逻辑完全对齐。

方案2：直接使用AsynchronousSocketChannel #

如果希望用原生NIO.2的异步API，直接按照官方示例写客户端代码即可，无需修改服务端，示例代码如下：

import java.net.InetSocketAddress
import java.nio.ByteBuffer
import java.nio.channels.AsynchronousSocketChannel
import java.nio.channels.CompletionHandler
import java.util.concurrent.TimeUnit

class Nio2TcpClient {
    private val address = InetSocketAddress("192.168.1.200", 33322)
    private val buffer = ByteBuffer.allocate(4096)
    private var clientChannel: AsynchronousSocketChannel? = null

    fun connect() {
        clientChannel = AsynchronousSocketChannel.open()
        clientChannel?.connect(address, null, object : CompletionHandler<Void, Void> {
            override fun completed(result: Void?, attachment: Void?) {
                // 连接成功，启动异步读取
                startReceive()
            }

            override fun failed(exc: Throwable?, attachment: Void?) {
                // 连接失败处理
            }
        })
    }

    private fun startReceive() {
        buffer.clear()
        clientChannel?.read(buffer, 5, TimeUnit.SECONDS, null, object : CompletionHandler<Int, Void> {
            override fun completed(readBytes: Int, attachment: Void?) {
                if (readBytes <= 0) {
                    disconnect()
                    return
                }
                buffer.flip()
                val data = ByteArray(readBytes)
                buffer.get(data)
                // 处理收到的数据
                onReceiveData(data)
                // 继续读取下一批数据
                startReceive()
            }

            override fun failed(exc: Throwable?, attachment: Void?) {
                // 读取异常处理
                disconnect()
            }
        })
    }

    private fun onReceiveData(data: ByteArray) {
        // 业务逻辑处理
    }

    fun send(data: ByteArray) {
        val writeBuffer = ByteBuffer.wrap(data)
        clientChannel?.write(writeBuffer, null, object : CompletionHandler<Int, Void> {
            override fun completed(result: Int?, attachment: Void?) {
                // 发送成功处理
            }

            override fun failed(exc: Throwable?, attachment: Void?) {
                // 发送失败处理
            }
        })
    }

    fun disconnect() {
        runCatching { clientChannel?.close() }
        clientChannel = null
    }
}

这个方案的回调模式和你给出的C#代码逻辑几乎完全一致，适配成本也很低。

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