如何在UDP协议上同步通信？

Question

我运行了一个模拟测试，以了解UDP通信面临的问题。

我有一个主机，其中有4个UDP客户机正在运行，每个客户机都有自己的线程（T1、T2、T3和T4）。T1和T2共享一个名为socket的DatagramSocket对象，而T3和T4共享一个名为socket2的DatagramSocket对象。

T1和T2正在发送和接收来自IP（例如）udpServer1的UDP服务器的回波，而T3和T4正在与udpServer2通信。

当并行运行线程时，我能够同步共享相同DatagramSocket和Runnable对象的T1和T2。然而，当尝试运行使用不同DatagramSocket和Runnable对象的T1和T3时，T3总是在SocketTimeoutExc0019下失败，但是T1可以毫无问题地发送和接收数据包。

总之：

>

并行运行T1和T2（或T3和T4），共享相同的DatagramSocket和Runnable对象--

并行运行T1和T3，每个都使用自己的DatagramSocket和Runnable对象--

为什么T3在自己的DatagramSocket上并在自己的线程上运行时，仍然会获得SocketTimeoutExcsion？我到底做错了什么？

任何帮助都非常感谢。谢谢。

@Test
public void sendRemoteTest() throws InterruptedException, IOException {
    DatagramSocket socket = new DatagramSocket(9987, InetAddress.getByName(localMachineIp));
    DatagramSocket socket2 = new DatagramSocket(9988, InetAddress.getByName(localMachineIp));
    UdpClientRunnable runnable = new UdpClientRunnable(socket, 1);
    UdpClientRunnable runnable2 = new UdpClientRunnable(socket2, 1);
   
    Thread t1 = new Thread(runnable);
    t1.setName("T1");
    Thread t2 = new Thread(runnable);
    t2.setName("T2");
    Thread t3 = new Thread(runnable2);
    t3.setName("T3");
    Thread t4 = new Thread(runnable2);
    t4.setName("T4");
    t1.start();
    t2.start();
    t3.start();
    t4.start();
    Thread.sleep(10000);
    t1.join();
    t2.join();
    t3.join();
    t4.join();
}

private byte toByte(int num) {
    return (byte) ((byte) (0xFF) & num);
}

public class UdpClientRunnable implements Runnable {
    private DatagramSocket socket;
    private long delayMillis;
    private byte[] dataToSend;

    public UdpClientRunnable(DatagramSocket socket, long delayMilis) {
       this.socket = socket;
       this.delayMillis = delayMilis;
    }

    @Override
    public synchronized void run() {
       byte[] data1 = new byte[] { toByte(0x01), toByte(0x01), toByte(0x01),
          toByte(0x01), toByte(0x01) };
       byte[] data2 = new byte[] { toByte(0x02), toByte(0x02), toByte(0x02),
          toByte(0x02), toByte(0x02) };
       byte[] data3 = new byte[] { toByte(0x03), toByte(0x03), toByte(0x03),
          toByte(0x03), toByte(0x03) };
       byte[] data4 = new byte[] { toByte(0x04), toByte(0x04), toByte(0x04),
          toByte(0x04), toByte(0x04) };

       String targetIp = "";
       String name = Thread.currentThread().getName();
       if (name.contains("T1")) {
          dataToSend = data1;
          targetIp = udpServer1;
       }
       else if (name.contains("T2")){
          dataToSend = data2;
          targetIp = udpServer1;
       }
       else if (name.contains("T3")) {
          dataToSend = data3;
          targetIp = udpServer2;
       }
       else {
          dataToSend = data4;
          targetIp = udpServer2;
       }
       int count = 0;
       while (count < 250) {
          try {
             sendAndReceive(targetIp, name, count, dataToSend);
             Thread.sleep(delayMillis);
          }
          catch (IOException | InterruptedException e) {
             System.out.println(e + ": " + name + ", iter: " + count);
          }
          finally {
             ++count;
          }
       }
    }

    private synchronized void sendAndReceive(String targetIp, String threadName, int count, byte[] dataToSend) throws IOException, UnknownHostException, SocketException {
        byte[] rcvData = new byte[5];
        DatagramPacket rcvPacket = new DatagramPacket(rcvData, rcvData.length);
        socket.send(new DatagramPacket(dataToSend, 5, InetAddress.getByName(targetIp), 9999));
        socket.setSoTimeout(2000);
        rcvPacket = new DatagramPacket(rcvData, rcvData.length);
        socket.receive(rcvPacket);
        printData(threadName, count, rcvData);
        if (threadName.contains("T1")) {
            Assert.assertArrayEquals(new byte[] { toByte(0x01), toByte(0x01), toByte(0x01),
                toByte(0x01), toByte(0x01) }, rcvData);
        }
        else if (threadName.contains("T2")){
            Assert.assertArrayEquals(new byte[] { toByte(0x02), toByte(0x02), toByte(0x02),
                toByte(0x02), toByte(0x02) }, rcvData);
        }
        else if (threadName.contains("T3")){
            Assert.assertArrayEquals(new byte[] { toByte(0x03), toByte(0x03), toByte(0x03),
                toByte(0x03), toByte(0x03) }, rcvData);
        }
        else if (threadName.contains("T4")){
            Assert.assertArrayEquals(new byte[] { toByte(0x04), toByte(0x04), toByte(0x04),
                toByte(0x04), toByte(0x04) }, rcvData);
        }
    }

    private void printData(String name, int count, byte[] rcvData) {
        String prefix = "";
        for (int i = 0; i < rcvData.length; i++) {
            prefix = (i == 0) ? (name + " iter " + count + ": ") : "";
            System.out.print(prefix + Integer.toHexString((byte) ((byte) (0xFF) & rcvData[i])) + " ");
        }
        System.out.println();
    }
  
}

Answer 1

查看代码，有两台服务器位于两台不同的机器上，我假定侦听的是同一端口999。因此，我认为问题可能在于服务器代码。当你回复客户时，你是这样做的吗

byte[] rcvData = new byte[5];
DatagramPacket rcvPacket = new DatagramPacket(rcvData, rcvData.length);
socket.receive(rcvPacket);
DatagramPacket sendPacket = new DatagramPacket(rcvPacket.getData(), rcvPacket.getLength(), rcvPacket.getAddress(), rcvPacket.getPort());  // note that here we need to use the received packet information on where to send the response back
socket.send(sendPacket);

很少有其他评论

1. 将构造函数UdpClientRunnable设置为附加输入目标端口和目标地址，这样就不需要在其中包含所有if-else块