Java DatagramChannel 挂起

Question

我正在开始一些 Java NIO 编程，并且正在构建两个独立的 Java 程序。一种是创建一些随机数据通过 UDP 发送，另一种是接收这些数据并可能对其进行一些处理。

在每个类中，我定义了要发送/接收的数据的大小，以及一次发送或接收的字节数（每个数据包）为 8 个字节。当我的 SIZE 为 1024 字节时，我可以打开接收器（默认情况下等待某些数据包到达），然后打开发送数据包的发送器，接收器正确接收这些数据包并在所有数据包结束时终止数据包被接收。然后我尝试将 SIZE 设置为 10 兆字节。我启动接收器，然后启动发送器。发件人似乎工作正常，几秒钟后终止。接收器开始工作并继续读取一些数据包（我在代码周围放置了一些 printlns）。然而，在这个过程的某个地方，接收器似乎挂起并停止接收更多的数据包。我不知道挂起发生在哪里，除了它发生在缓冲区被清除以进行新读取之后。

也许我错误地使用了库。非常感谢任何帮助。

发送者类称为UDPSender，如下所示：

    public class UDPSender {

    public static void main(String[] args) {
        try {
            // Generate some data to send
            int SIZE=1024*1024*10;
            int bufferBytes = 8;

            byte[] b = new byte[SIZE];
            new Random().nextBytes(b);
            String value = new String(b);
            //System.out.println(value);

            // Create Datagram channel for sending UDP packets
            DatagramChannel sendingChannel = DatagramChannel.open();
            // Bind the channel to a specific sending socket address and port
            InetSocketAddress sendSocket = new InetSocketAddress("127.0.0.1", 9999);
            sendingChannel.socket().bind(sendSocket);

            //Target socket address
            InetSocketAddress targetSocket = new InetSocketAddress("127.0.0.1", 8888);
            //loop to send bufferBytes bytes at a time
            int location = 0;
            int bytesSent = 0;
            long startTime=0; long endTime=0; long elapsedTime=0;  // for speed benchmarking
            startTime = System.nanoTime();
            for (int i = 1; i <= SIZE / bufferBytes; i++) {
                //System.out.println(i);
                ByteBuffer buf = ByteBuffer.allocate(bufferBytes);
                buf.clear();
                buf.put(Arrays.copyOfRange(b, location, location + bufferBytes));
                location = location + bufferBytes;
                buf.flip();
                bytesSent = bytesSent + sendingChannel.send(buf, targetSocket);
            }
            endTime = System.nanoTime();
            elapsedTime = endTime - startTime;
            double elapsedTimeMs = elapsedTime / 1000000.0;
            System.out.println("Elapsed:" + elapsedTimeMs + " msec");
        } catch (IOException ex) {
            Logger.getLogger(UDPSender.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
}

第二个类叫做UDPReceiver，如下图：

    public class UDPReceiver {
    public static void main(String[] args) {
        try {
            int SIZE=1024*1024*10;
            int bufferBytes = 8;
            int packet_max = SIZE/bufferBytes;

            // Create Datagram channel for receiving UDP packets
            DatagramChannel receivingChannel = DatagramChannel.open();
            // Bind the channel to a specific receiving socket address and port
            InetSocketAddress receiveSocket = new InetSocketAddress("127.0.0.1",8888);
            receivingChannel.socket().bind(receiveSocket);
            // Set up address of send
            InetSocketAddress sendSocket = new InetSocketAddress("127.0.0.1",9999);
            // Connect our receiving socket to their sending socket address
            receivingChannel.connect(sendSocket);

            long startTime=0; long endTime=0; long elapsedTime=0;  // for speed benchmarking
            byte[] b = new byte[SIZE]; //receiving storage
            int packets=0; int location=0; int bytesCount; //byte[] indexes
            boolean clockStart=false;


            //try indirect ByteBuffer
            System.out.println("Indirect ByteBuffer()");
            ByteBuffer buf = ByteBuffer.allocate(bufferBytes);
            packets=0; location=0; bytesCount = 0; clockStart=false;
            while ((bytesCount = receivingChannel.read(buf)) > 0) { // Read data from file into ByteBuffer
                System.out.println(".");
                if(clockStart==false){
                    startTime = System.nanoTime();
                    clockStart=true;
                }
                // flip the buffer which set the limit to current position, and position to 0.
                buf.flip(); System.out.println("..");
                System.arraycopy(buf.array(), 0, b, location, buf.capacity()); System.out.println("..."); // Write data from ByteBuffer to bytearray
                location=location+bufferBytes; System.out.println("....");
                buf.clear();  System.out.println(".....");    // For the next read
                packets = packets+1; System.out.println("......");
                if(packets==packet_max){
                    System.out.println(".......");
                    System.out.println(packets);
                    break;
                }
            }
            endTime = System.nanoTime();
            elapsedTime = endTime - startTime;
            System.out.println("ByteBuffer() Elapsed Time is " + (elapsedTime / 1000000.0) + " msec");

        } catch (IOException ex) {
            System.out.println(ex.toString());
            //Logger.getLogger(UDPReceiver.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
}

Answer 1

它不会“挂起”，它阻塞， 这样做是因为您试图读取特定数量的数据包，而不考虑 UDP 不可靠的事实。当您丢失数据包时，您的发送者和接收者都不知道，因此您的发送者不会重新发送，并且您的接收者会继续尝试读取全部数量的数据包，而这些数据包永远不会到达。