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

Question

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

代码在这里：

在一个线程中接收数据包并提供给一个LinkedBlockingQueue线程，在另一个线程上消费从一个 rx-java 主题中获取数据包LinkedBlockingQueue，然后调用onNext它。

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

Answer 1

我最近在另一种语言中遇到了类似的问题。我不确定它在 Java 中是否同样有效，但这可能对您有所帮助。因此，当数据包进入套接字时，它们被缓冲并且您已经设置了缓冲区大小，但是您仍然只读取一个数据包，即使缓冲区可能容纳更多。当您一次处理一个数据报时，您的缓冲区会越来越多，最终当它满时，数据可能会丢失，因为它无法存储更多数据报。我检查了文档的DatagramSocketReceives a datagram packet from this socket我不确定您需要在 Java 中调用哪些函数，但这里有一个我正在使用的小片段。while (!m_server->BufferEmpty()) {&nbsp; &nbsp; &nbsp; &nbsp; std::shared_ptr<Stream> inStream = std::make_shared<Stream>();&nbsp; &nbsp; &nbsp; &nbsp; std::vector<unsigned char>& buffer = inStream->GetBuffer();&nbsp; &nbsp; &nbsp; &nbsp; boost::asio::ip::udp::endpoint senderEndpoint = m_server->receive(boost::asio::buffer(buffer),&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; boost::posix_time::milliseconds(-1), ec);&nbsp; &nbsp; &nbsp; &nbsp; if (ec)&nbsp; &nbsp; &nbsp; &nbsp; {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; std::cout << "Receive error: " << ec.message() << "\n";&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; else&nbsp; &nbsp; &nbsp; &nbsp; {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; std::unique_ptr<IPacketIn> incomingPacket = std::make_unique<IPacketIn>();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; incomingPacket->ReadHeader(inStream);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; m_packetProcessor->ProcessPacket(incomingPacket, senderEndpoint);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; incomingPacket.reset();&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; ++packetsRead;&nbsp; &nbsp; &nbsp; &nbsp; inStream.reset();}这基本上是说，如果套接字在其缓冲区中有当前帧的任何数据，则继续读取数据报，直到缓冲区为空。不确定它是如何LinkedBlockingQueue工作的，但是如果两个线程都试图同时访问它，这也可能会导致一些问题。在您的 UDP 读取线程中，您可能会被阻塞一段时间，然后在此期间可能会收到数据包。

Answer 2

很难给出一个直接的答案，但根据我在 Java 中处理 UDP 消息的经验，提高处理消息的性能确实很重要，尤其是处理大量数据时。所以这里有一些我会考虑的事情：1) 在不同的队列上处理 UDP 消息是正确的。但是，队列的大小有限。您是否设法快速处理消息？否则，队列会填满，而您将阻塞 while 循环。如果是这种情况，一些简单的日志记录可以让您知道。把它们放在一个队列中，在那里它们可以在不同的步骤中被排出是很棒的，但你还需要确保处理速度很快并且队列不会填满。2) 你所有的数据报都小于 532 字节吗？可能由于未填充缓冲区的较大消息而发生一些损失。希望这可以帮助，