C#實現UDP分包組包

本文為 Dennis Gao 原創技術文章，發表於博客園博客，未經作者本人允許禁止任何形式的轉載。

場景介紹

如果需要使用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);
      }
    }

本文為 Dennis Gao 原創技術文章，發表於博客園博客，未經作者本人允許禁止任何形式的轉載。