C# UDP 大数据分包传输

如果需要使用UDP传输较大数据，例如传输10M的图片，这突破了UDP的设计原则。UDP的设计是基于"datagram"，也就是它假设你发送的每个数据包都能包含在单一的包内。并且设定UDP数据包的最大长度受基础网络协议的限制。

UDP数据包的理论最大长度限制是 65535 bytes，这包含 8 bytes 数据包头和 65527 bytes 数据。但如果基于IPv4网络传输，则还需减去 20 bytes 的IP数据包头。
则单一的UDP数据包可传输的数据最大长度为：

MaxUdpDataLength = 65535 - 8 - 20 = 65507 bytes

这就需要实现UDP包的分包传输和接收组包功能。

 

分包功能

/// <summary>
/// UDP数据包分割器
/// </summary>
public static class UdpPacketSplitter
{
  /// <summary>
  /// 分割UDP数据包
  /// </summary>
  /// <param name="sequence">UDP数据包所持有的序号</param>
  /// <param name="datagram">被分割的UDP数据包</param>
  /// <param name="chunkLength">分割块的长度</param>
  /// <returns>
  /// 分割后的UDP数据包列表
  /// </returns>
  public static ICollection<UdpPacket> Split(long sequence, byte[] datagram, int chunkLength)
  {
    if (datagram == null)
      throw new ArgumentNullException("datagram");

    List<UdpPacket> packets = new List<UdpPacket>();

    int chunks = datagram.Length / chunkLength;
    int remainder = datagram.Length % chunkLength;
    int total = chunks;
    if (remainder > 0) total++;

    for (int i = 1; i <= chunks; i++)
    {
      byte[] chunk = new byte[chunkLength];
      Buffer.BlockCopy(datagram, (i - 1) * chunkLength, chunk, 0, chunkLength);
      packets.Add(new UdpPacket(sequence, total, i, chunk, chunkLength));
    }
    if (remainder > 0)
    {
      int length = datagram.Length - (chunkLength * chunks);
      byte[] chunk = new byte[length];
      Buffer.BlockCopy(datagram, chunkLength * chunks, chunk, 0, length);
      packets.Add(new UdpPacket(sequence, total, total, chunk, length));
    }

    return packets;
  }
}

..............................

发送分包

..............................

private void WorkThread()
{
  while (IsRunning)
  {
    waiter.WaitOne();
    waiter.Reset();

    while (queue.Count > 0)
    {
      StreamPacket packet = null;
      if (queue.TryDequeue(out packet))
      {
        RtpPacket rtpPacket = RtpPacket.FromImage(
          RtpPayloadType.JPEG, 
          packet.SequenceNumber, 
          (long)Epoch.GetDateTimeTotalMillisecondsByYesterday(packet.Timestamp),
          packet.Frame);

        // max UDP packet length limited to 65,535 bytes
        byte[] datagram = rtpPacket.ToArray(); 
        packet.Frame.Dispose();

        // split udp packet to many packets 
        // to reduce the size to 65507 limit by underlying IPv4 protocol
        ICollection<UdpPacket> udpPackets 
          = UdpPacketSplitter.Split(
            packet.SequenceNumber, 
            datagram, 
            65507 - UdpPacket.HeaderSize);
        foreach (var udpPacket in udpPackets)
        {
          byte[] udpPacketDatagram = udpPacket.ToArray();
          // async sending
          udpClient.BeginSend(
            udpPacketDatagram, udpPacketDatagram.Length,
            packet.Destination.Address,
            packet.Destination.Port,
            SendCompleted, udpClient);
        }
      }
    }
  }
}

接收组包功能

..............................

private void OnDatagramReceived(object sender, UdpDatagramReceivedEventArgs<byte[]> e)
{
  try
  {
    UdpPacket udpPacket = UdpPacket.FromArray(e.Datagram);

    if (udpPacket.Total == 1)
    {
      RtpPacket packet = new RtpPacket(udpPacket.Payload, udpPacket.PayloadSize);
      Bitmap bitmap = packet.ToBitmap();
      RaiseNewFrameEvent(
        bitmap, Epoch.GetDateTimeByYesterdayTotalMilliseconds(packet.Timestamp));
    }
    else
    {
      // rearrange packets to one packet
      if (packetCache.ContainsKey(udpPacket.Sequence))
      {
        List<UdpPacket> udpPackets = null;
        if (packetCache.TryGetValue(udpPacket.Sequence, out udpPackets))
        {
          udpPackets.Add(udpPacket);

          if (udpPackets.Count == udpPacket.Total)
          {
            packetCache.TryRemove(udpPacket.Sequence, out udpPackets);

            udpPackets = udpPackets.OrderBy(u => u.Order).ToList();
            int rtpPacketLength = udpPackets.Sum(u => u.PayloadSize);
            int maxPacketLength = udpPackets.Select(u => u.PayloadSize).Max();

            byte[] rtpPacket = new byte[rtpPacketLength];
            foreach (var item in udpPackets)
            {
              Buffer.BlockCopy(
                item.Payload, 0, rtpPacket, 
                (item.Order - 1) * maxPacketLength, item.PayloadSize);
            }

            RtpPacket packet = new RtpPacket(rtpPacket, rtpPacket.Length);
            Bitmap bitmap = packet.ToBitmap();
            RaiseNewFrameEvent(
              bitmap, 
              Epoch.GetDateTimeByYesterdayTotalMilliseconds(packet.Timestamp));

            packetCache.Clear();
          }
        }
      }
      else
      {
        List<UdpPacket> udpPackets = new List<UdpPacket>();
        udpPackets.Add(udpPacket);
        packetCache.AddOrUpdate(
          udpPacket.Sequence, 
          udpPackets, (k, v) => { return udpPackets; });
      }
    }
  }
  catch (Exception ex)
  {
    RaiseVideoSourceExceptionEvent(ex.Message);
  }
}

.......................

 

 

 

...