为啥 Java DatagramSocket 没有收到客户端发送的所有 udp 数据包？

Posted 2023-02-24

【中文标题】为啥 Java DatagramSocket 没有收到客户端发送的所有 udp 数据包？

【英文标题】：Why Java DatagramSocket doesn't receive all udp packet that clients have sent?为什么 Java DatagramSocket 没有收到客户端发送的所有 udp 数据包？

【发布时间】：2018-08-27 05:59:15

【问题描述】：

我的客户以高速率发送 udp 数据包。 我确定我的 java 应用程序层没有收到客户端发送的所有 udp 数据包，因为在 wireshark 中收到的数据包数量和我的 java 应用程序不匹配。 因为wireshark接收到更多的udp数据包，所以我确定udp数据包没有在网络中丢失。

代码在这里：

在一个线程中接收数据包并提供给LinkedBlockingQueue，然后在另一个线程上使用来自LinkedBlockingQueue 的数据包，然后调用onNext rx-java 主题。

socket = new DatagramSocket(this.port);
socket.setBroadcast(true);
socket.setReceiveBufferSize(2 * 1024 * 1024);

// thread-1
while (true) 
  byte[] bytes = new byte[532];
  DatagramPacket packet = new DatagramPacket(bytes, bytes.length);
  try 
    this.socket.receive(packet);
    queue.offer(
        new UdpPacket(
            packet.getPort(), packet.getAddress().getHostAddress(), packet.getData()));
   catch (IOException e) 
    e.printStackTrace();
  



// thread-2
UdpPacket packet;
while ((packet = queue.take()) != null) 
     this.receiveMessageSubject.onNext(packet);

主机操作系统：Ubutnu 18.04

【参考方案1】：

很难给出直接的答案，但根据我在 Java 中处理 UDP 消息的经验，提高处理消息的性能确实很重要，尤其是在处理大量数据的情况下。

所以这里有一些我会考虑的事情：

1) 您在不同队列上处理 UDP 消息是正确的。但是，队列的大小是有限的。您是否设法快速处理消息？否则，队列已满，您将阻塞 while 循环。如果是这种情况，一些简单的日志记录可以让您知道。将它们放在可以在不同步骤中排出的队列中很棒，但您还需要确保处理速度尽可能快并且队列不会填满。

2) 你所有的数据报都小于 532 字节吗？可能由于没有填满缓冲区的较大消息而发生一些丢失。

希望这会有所帮助，

【参考方案2】：

我最近用另一种语言遇到了类似的问题。我不确定它在 Java 中是否同样有效，但这可能对您有所帮助。

因此，当数据包进入套接字时，它们会被缓冲，并且您已经设置了缓冲区大小，但您仍然只读取单个数据包，即使缓冲区可能容纳更多。当您一次处理一个数据报时，您的缓冲区会填得更多，最终当其满时，数据可能会丢失，因为它无法存储更多数据报。

我检查了documentation 的DatagramSocket

Receives a datagram packet from this socket

我不确定您需要在 Java 中调用哪些函数，但这里有一个我正在使用的小 sn-p。

while (!m_server->BufferEmpty()) 
        std::shared_ptr<Stream> inStream = std::make_shared<Stream>();
        std::vector<unsigned char>& buffer = inStream->GetBuffer();

        boost::asio::ip::udp::endpoint senderEndpoint = m_server->receive(boost::asio::buffer(buffer),
            boost::posix_time::milliseconds(-1), ec);

        if (ec)
        
            std::cout << "Receive error: " << ec.message() << "\n";
        
        else
        
            std::unique_ptr<IPacketIn> incomingPacket = std::make_unique<IPacketIn>();
            incomingPacket->ReadHeader(inStream);

            m_packetProcessor->ProcessPacket(incomingPacket, senderEndpoint);

            incomingPacket.reset();
        

        ++packetsRead;

        inStream.reset();

这基本上是说，如果套接字在其缓冲区中有当前帧的任何数据，则继续读取数据报，直到缓冲区为空。

不确定LinkedBlockingQueue 是如何工作的，但是如果两个线程同时尝试访问它，这也可能会导致一些问题。在您的 UDP 读取线程中，您可能会被阻塞一段时间，然后在此期间可能会收到数据包。