C#异步TCP服务器多客户端JSON数据有序处理问题咨询

我正在学习C#，开发了一个支持多客户端连接的异步TCP服务器，但在将数据稳定传入processData(String)方法时遇到问题。经排查，问题出在字节缓冲区：当多个客户端发送数据，或单个客户端连续快速发送数据时，会出现数据拼接、不完整的情况。

我尝试过以下方案，但都没彻底解决：

• 用isWorking布尔值防止processData()被并发访问，却引发了JSONReader Exception，推测是传入了不完整字符串。
• 在字符串末尾添加[EOF]标记，检测到标记时调用processData()，但缓冲区出现半段字符串加另一段开头的问题。
• 尝试用}作为分隔符拆分响应，代码如下：

string data = content.Replace("[EOF]", "");
string[] TooManyDataObjects = data.Split('}');
string firstObject = TooManyDataObjects[0] += "}";

依然存在半段字符串加另一段开头的问题。

以下是创建监听任务及设备连接时HandleDevice()方法的代码：

private async Task Listen()
{
    try
    {
        while (true)
        {
            Token.ThrowIfCancellationRequested();
            TcpClient client = await server.AcceptTcpClientAsync();
            client.NoDelay = true;
            Console.WriteLine("Connected!");
            await Task.Run(async () => await HandleDevice(client), Token);
        }
    }
    catch (SocketException e)
    {
        Console.WriteLine("Exception: {0}", e);
    }
}

private async Task HandleDevice(TcpClient client)
{
    string data = null;
    Byte[] bytes = new Byte[8196];
    int i;

    try
    {
        using (stream = client.GetStream())
        {
            while ((i = await stream.ReadAsync(bytes, 0, bytes.Length, Token)) != 0)
            {
                Token.ThrowIfCancellationRequested();
                string hex = BitConverter.ToString(bytes);
                data += Encoding.UTF8.GetString(bytes, 0, i);

                //Console.WriteLine(data);
                if (data.IndexOf("[EOF]") >= 0)
                {
                    processData(data);
                    data = "";
                    Array.Clear(bytes, 0, bytes.Length);
                }
          
            }
        }
    }
    catch (OperationCanceledException) { }
    catch (Exception e)
    {
        Console.WriteLine("Exception: {0}", e.ToString());
    }
    finally
    {
        client.Close();
    }
}

核心诉求：如何确保从客户端接收的格式正确的JSON字符串能按接收顺序或队列形式传入processData()方法？

1. 正确处理[EOF]分隔符，拆分完整消息 #

你的[EOF]思路是对的，但之前的实现只是简单判断存在就处理整个缓冲区，没考虑缓冲区可能包含多个完整消息+半段消息的情况。正确的处理逻辑应该是：

• 每次读取数据后，循环检查缓冲区中是否有[EOF]标记
• 找到标记后，拆分出完整的消息段传入processData()
• 保留剩余的半段消息到下一次循环处理

修改后的HandleDevice核心逻辑：

private async Task HandleDevice(TcpClient client)
{
    var buffer = new StringBuilder(); // 用StringBuilder优化字符串拼接性能
    Byte[] bytes = new Byte[8196];
    int i;
    const string EofMarker = "[EOF]";

    try
    {
        using (var stream = client.GetStream()) // 不要用全局stream，每个客户端单独持有流实例
        {
            while ((i = await stream.ReadAsync(bytes, 0, bytes.Length, Token)) != 0)
            {
                Token.ThrowIfCancellationRequested();
                buffer.Append(Encoding.UTF8.GetString(bytes, 0, i));
                string currentData = buffer.ToString();

                int eofIndex;
                // 循环处理所有完整的消息
                while ((eofIndex = currentData.IndexOf(EofMarker)) != -1)
                {
                    // 提取完整消息（去掉[EOF]标记）
                    string completeMessage = currentData.Substring(0, eofIndex);
                    processData(completeMessage);

                    // 保留剩余的半段消息，用于下一次拼接
                    currentData = currentData.Substring(eofIndex + EofMarker.Length);
                }

                // 更新缓冲区为剩余的半段消息
                buffer.Clear();
                buffer.Append(currentData);
                Array.Clear(bytes, 0, bytes.Length);
            }
        }
    }
    catch (OperationCanceledException) { }
    catch (Exception e)
    {
        Console.WriteLine("Exception: {0}", e.ToString());
    }
    finally
    {
        client.Close();
    }
}

2. 保证processData()的线程安全与顺序执行 #

如果多个客户端的消息需要严格按接收顺序处理，或者processData()本身不是线程安全的，需要引入消息队列+单线程处理机制：

• 用ConcurrentQueue<string>作为全局消息队列，所有HandleDevice线程只负责将完整消息入队
• 单独启动一个后台线程，循环从队列中取出消息调用processData()

示例实现：

// 全局消息队列与通知事件
private readonly ConcurrentQueue<string> _messageQueue = new ConcurrentQueue<string>();
private readonly AutoResetEvent _queueEvent = new AutoResetEvent(false);

// 服务器启动时启动处理线程
public void StartMessageProcessor()
{
    Task.Run(() =>
    {
        while (!Token.IsCancellationRequested)
        {
            _queueEvent.WaitOne(); // 等待队列有新消息
            // 一次性处理队列中所有消息
            while (_messageQueue.TryDequeue(out string message))
            {
                processData(message);
            }
        }
    }, Token);
}

// 修改HandleDevice中的处理逻辑，将消息入队而非直接调用processData
// ...
string completeMessage = currentData.Substring(0, eofIndex);
_messageQueue.Enqueue(completeMessage);
_queueEvent.Set(); // 通知处理线程有新消息
// ...

3. 可选：用JSON长度前缀替代分隔符（更可靠） #

如果客户端可以配合修改通信协议，推荐用长度前缀的方式（比分隔符更可靠，避免JSON内容中意外包含[EOF]的情况）：

• 客户端先发送4字节的Int32（表示后续JSON的字节长度），再发送JSON内容
• 服务器先读取长度前缀，再读取对应长度的字节，确保拿到完整的JSON

示例读取逻辑：

// 先读取4字节的长度前缀
byte[] lengthBuffer = new byte[4];
await stream.ReadAsync(lengthBuffer, 0, 4, Token);
int jsonByteLength = BitConverter.ToInt32(lengthBuffer, 0);

// 再读取对应长度的JSON字节
byte[] jsonBuffer = new byte[jsonByteLength];
int totalRead = 0;
// 循环读取直到拿到完整的JSON字节
while (totalRead < jsonByteLength)
{
    int readBytes = await stream.ReadAsync(jsonBuffer, totalRead, jsonByteLength - totalRead, Token);
    totalRead += readBytes;
}

string completeJson = Encoding.UTF8.GetString(jsonBuffer);
processData(completeJson);

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