如果需要使用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 bytes20 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 protocol25 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 sending34 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 else15 {16 // rearrange packets to one packet17 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 else51 {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 }
联系客服