场景介绍
如果需要使用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 /// <summary> 2 /// UDP数据包分割器 3 /// </summary> 4 public static class UdpPacketSplitter 5 { 6 /// <summary> 7 /// 分割UDP数据包 8 /// </summary> 9 /// <param name="sequence">UDP数据包所持有的序号</param> 10 /// <param name="datagram">被分割的UDP数据包</param> 11 /// <param name="chunkLength">分割块的长度</param> 12 /// <returns> 13 /// 分割后的UDP数据包列表 14 /// </returns> 15 public static ICollection<UdpPacket> Split(long sequence, byte[] datagram, int chunkLength) 16 { 17 if (datagram == null) 18 throw new ArgumentNullException("datagram"); 19 20 List<UdpPacket> packets = new List<UdpPacket>(); 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<UdpPacket> 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<byte[]> 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<UdpPacket> 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<UdpPacket> udpPackets = new List<UdpPacket>(); 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 }
