921600波特率下RS-422实时接收字节数组丢包问题求助

问题场景 #

通过RS-422以921600波特率实时接收字节数组，单次接收约3500字节，但后续会丢失约2800字节。接收逻辑通过client.OnReceiving += new EventHandler<DataStreamEventArgs>(receiveHandler);绑定事件回调，核心代码如下：

原接收处理代码 #

void receiveHandler(object sender, DataStreamEventArgs e)
{
  ReadData(e.Response);
}

void ReadData (byte[] byteBuffer)
{
  if (byteBuffer != null)
  {
    for (int i = 0; i < byteBuffer.Length; i++)
    {
      if (i < byteBuffer.Length-1 && byteBuffer[i] == 0xAA && byteBuffer[++i] == 0xBA)
      {
        results = new byte[138];
        Buffer.BlockCopy(byteBuffer, i + 1, results, 0, results.Length);
        byte[] h = { results[results.Length - 2], results[results.Length - 1] };
        ControlSum = (ushort)BitConverter.ToInt16(h, 0);
        results = results.Take(results.Count() - 2).ToArray();
        ushort u = checkSum.CRC_Calc(results);
        PackCounterForAll++;
        if (ControlSum == u)
        {
          int offset = 0;
          @float = results.Skip(8)
            .ToArray()
            .GroupBy(x => offset++ / 4)
            .Select(x => BitConverter.ToSingle(x.ToArray(), 0))
            .ToArray();
          PackCounter++;
        }
      }
    }
  }
}

底层接收方法 #

public int Receive(byte[] bytes, int offset, int count)
{
    int readBytes = 0;

    if (count > 0)
    {
        readBytes = _serialPort.Read(bytes, offset, count);
    }

    return readBytes;
}

问题根源分析 #

1. 回调线程阻塞：在OnReceiving回调中直接执行CRC计算、浮点转换等耗时操作，导致串口接收缓冲区无法及时读取新数据，溢出丢包。
2. 跨缓冲区数据包丢失：当前解析逻辑仅处理单块缓冲区中的数据包，若帧头0xAA 0xBA跨两个接收缓冲区，会直接丢失该数据包。
3. 索引操作错误：原循环中找到帧头后使用++i，跳过了下一个字节的检查，若存在连续帧头会丢失后续数据。
4. 串口缓冲区过小：默认的串口接收缓冲区大小（通常1024字节）无法匹配921600波特率的高速传输，导致数据溢出。

解决措施 #

1. 异步解包，避免回调阻塞 #

将接收回调简化为仅缓存数据，使用独立工作线程处理解包逻辑，确保串口接收线程不被阻塞：

using System.Collections.Concurrent;
using System.Threading;

private readonly ConcurrentQueue<byte[]> _dataQueue = new ConcurrentQueue<byte[]>();
private readonly ManualResetEventSlim _dataSignal = new ManualResetEventSlim(false);
private Task _processingTask;

// 初始化时启动工作线程
public void InitReceiver()
{
    _processingTask = Task.Run(ProcessDataQueue);
    client.OnReceiving += receiveHandler;
}

void receiveHandler(object sender, DataStreamEventArgs e)
{
    if (e.Response != null && e.Response.Length > 0)
    {
        _dataQueue.Enqueue(e.Response);
        _dataSignal.Set();
    }
}

void ProcessDataQueue()
{
    while (true)
    {
        _dataSignal.Wait();
        while (_dataQueue.TryDequeue(out var buffer))
        {
            ReadData(buffer);
        }
        _dataSignal.Reset();
    }
}

2. 维护全局缓冲区，处理跨块数据包 #

使用全局缓冲区整合所有接收数据，循环查找完整数据包，避免跨缓冲区丢失：

private readonly List<byte> _globalBuffer = new List<byte>();
private readonly object _bufferLock = new object();

void ReadData(byte[] byteBuffer)
{
    lock (_bufferLock)
    {
        _globalBuffer.AddRange(byteBuffer);
        // 循环处理所有完整数据包
        while (_globalBuffer.Count >= 140) // 2字节帧头 + 138字节数据包
        {
            int frameStartIndex = -1;
            // 查找帧头位置
            for (int i = 0; i < _globalBuffer.Count - 1; i++)
            {
                if (_globalBuffer[i] == 0xAA && _globalBuffer[i + 1] == 0xBA)
                {
                    frameStartIndex = i;
                    break;
                }
            }
            if (frameStartIndex == -1)
            {
                // 无有效帧头，清空无效数据（保留最后1字节防止帧头跨块）
                _globalBuffer.RemoveRange(0, _globalBuffer.Count - 1);
                break;
            }
            if (_globalBuffer.Count < frameStartIndex + 2 + 138)
            {
                // 数据不足完整包，等待下一批数据
                break;
            }
            // 提取完整数据包
            byte[] packet = new byte[138];
            _globalBuffer.CopyTo(frameStartIndex + 2, packet, 0, 138);
            // 校验与处理
            ushort controlSum = BitConverter.ToUInt16(new[] { packet[136], packet[137] }, 0);
            byte[] dataToCheck = packet.Take(136).ToArray();
            ushort calculatedCrc = checkSum.CRC_Calc(dataToCheck);
            PackCounterForAll++;
            if (controlSum == calculatedCrc)
            {
                int offset = 0;
                float[] floatData = dataToCheck.Skip(8)
                    .ToArray()
                    .GroupBy(x => offset++ / 4)
                    .Select(x => BitConverter.ToSingle(x.ToArray(), 0))
                    .ToArray();
                PackCounter++;
                // 在这里处理转换后的浮点数据（如更新UI、存储等）
            }
            // 移除已处理的数据
            _globalBuffer.RemoveRange(0, frameStartIndex + 2 + 138);
        }
        // 限制缓冲区最大长度，防止内存溢出
        if (_globalBuffer.Count > 139)
        {
            _globalBuffer.RemoveRange(0, _globalBuffer.Count - 139);
        }
    }
}

3. 增大串口接收缓冲区 #

初始化SerialPort时，将接收缓冲区调整为适合高速传输的大小：

_serialPort.ReadBufferSize = 8192; // 可根据实际情况调整为16384等更大值

4. 修复索引错误 #

原代码中++i会跳过后续字节检查，改用正常索引递增逻辑，避免遗漏连续帧头。

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