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包的分包传输和接收组包功能。

 1   /// 

 2   /// UDP数据包分割器
 3   /// 
 4   public static class UdpPacketSplitter
 5   {
 6     /// 

 7     /// 分割UDP数据包
 8     /// 
 9     /// UDP数据包所持有的序号
10     /// 被分割的UDP数据包
11     /// 分割块的长度
12     /// 
13     /// 分割后的UDP数据包列表
14     /// 
15     public static ICollection Split(long sequence, byte[] datagram, int chunkLength)
16     {
17       if (datagram == null)
18         throw new ArgumentNullException("datagram");
19 
20       List packets = new List();
21 
22       int chunks = datagram.Length / chunkLength;
23       int remainder = datagram.Length % chunkLength;
24       int total = chunks;
25       if (remainder > 0) total++;
26 
27       for (int i = 1; i <= chunks; i++)
28       {
29         byte[] chunk = new byte[chunkLength];
30         Buffer.BlockCopy(datagram, (i - 1) * chunkLength, chunk, 0, chunkLength);
31         packets.Add(new UdpPacket(sequence, total, i, chunk, chunkLength));
32       }
33       if (remainder > 0)
34       {
35         int length = datagram.Length - (chunkLength * chunks);
36         byte[] chunk = new byte[length];
37         Buffer.BlockCopy(datagram, chunkLength * chunks, chunk, 0, length);
38         packets.Add(new UdpPacket(sequence, total, total, chunk, length));
39       }
40 
41       return packets;
42     }
43   }

 1 private void WorkThread()
 2 {
 3   while (IsRunning)
 4   {
 5     waiter.WaitOne();
 6     waiter.Reset();
 7 
 8     while (queue.Count > 0)
 9     {
10       StreamPacket packet = null;
11       if (queue.TryDequeue(out packet))
12       {
13         RtpPacket rtpPacket = RtpPacket.FromImage(
14           RtpPayloadType.JPEG, 
15           packet.SequenceNumber, 
16           (long)Epoch.GetDateTimeTotalMillisecondsByYesterday(packet.Timestamp),
17           packet.frame);
18 
19         // max UDP packet length limited to 65,535 bytes
20         byte[] datagram = rtpPacket.ToArray(); 
21         packet.frame.Dispose();
22 
23         // split udp packet to many packets 
24         // to reduce the size to 65507 limit by underlying IPv4 protocol
25         ICollection udpPackets 
26           = UdpPacketSplitter.Split(
27             packet.SequenceNumber, 
28             datagram, 
29             65507 - UdpPacket.HeaderSize);
30         foreach (var udpPacket in udpPackets)
31         {
32           byte[] udpPacketDatagram = udpPacket.ToArray();
33           // async sending
34           udpClient.BeginSend(
35             udpPacketDatagram, udpPacketDatagram.Length,
36             packet.Destination.Address,
37             packet.Destination.Port,
38             SendCompleted, udpClient);
39         }
40       }
41     }
42   }
43 }

 1     private void onDatagramReceived(object sender, UdpDatagramReceivedEventArgs e)
 2     {
 3       try
 4       {
 5         UdpPacket udpPacket = UdpPacket.FromArray(e.Datagram);
 6 
 7         if (udpPacket.Total == 1)
 8         {
 9           RtpPacket packet = new RtpPacket(udpPacket.Payload, udpPacket.PayloadSize);
10           Bitmap bitmap = packet.ToBitmap();
11           RaiseNewframeEvent(
12             bitmap, Epoch.GetDateTimeByYesterdayTotalMilliseconds(packet.Timestamp));
13         }
14         else
15         {
16           // rearrange packets to one packet
17           if (packetCache.ContainsKey(udpPacket.Sequence))
18           {
19             List udpPackets = null;
20             if (packetCache.TryGetValue(udpPacket.Sequence, out udpPackets))
21             {
22               udpPackets.Add(udpPacket);
23 
24               if (udpPackets.Count == udpPacket.Total)
25               {
26                 packetCache.TryRemove(udpPacket.Sequence, out udpPackets);
27 
28                 udpPackets = udpPackets.OrderBy(u => u.Order).ToList();
29                 int rtpPacketLength = udpPackets.Sum(u => u.PayloadSize);
30                 int maxPacketLength = udpPackets.Select(u => u.PayloadSize).Max();
31 
32                 byte[] rtpPacket = new byte[rtpPacketLength];
33                 foreach (var item in udpPackets)
34                 {
35                   Buffer.BlockCopy(
36                     item.Payload, 0, rtpPacket, 
37                     (item.Order - 1) * maxPacketLength, item.PayloadSize);
38                 }
39 
40                 RtpPacket packet = new RtpPacket(rtpPacket, rtpPacket.Length);
41                 Bitmap bitmap = packet.ToBitmap();
42                 RaiseNewframeEvent(
43                   bitmap, 
44                   Epoch.GetDateTimeByYesterdayTotalMilliseconds(packet.Timestamp));
45 
46                 packetCache.Clear();
47               }
48             }
49           }
50           else
51           {
52             List udpPackets = new List();
53             udpPackets.Add(udpPacket);
54             packetCache.AddOrUpdate(
55               udpPacket.Sequence, 
56               udpPackets, (k, v) => { return udpPackets; });
57           }
58         }
59       }
60       catch (Exception ex)
61       {
62         RaiseVideoSourceExceptionEvent(ex.Message);
63       }
64     }