如何应用C#实现UDP的分包组包

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

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

则单一的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);
       }
     }