java - 带有序列号的 UDP

Question

我正在尝试为 Java 中的类分配实现可靠的 UDP 协议(protocol)。我已设法将确认添加到收到的每个数据报包中，但在我发送的数据报包中实现序列号时遇到问题。

谁能建议一个简单的方法来实现这个？

@EJP 我已尝试实现您刚才建议的内容。这是我到目前为止的代码(它仍然非常原始 - 我正在使用 hit 和 try 方法来实现它)

服务器端

    public class TestServer extends Activity {

private DatagramSocket serverSocket;
Thread serverThread = null;
byte[] incomingData;
byte[] outgoingData;
//int numBytesRead = 0;
int ackSent = 0;
int numPackRecv = 0;
int BUF_SIZE = 1024;
String msg = "ACK";
BufferedInputStream data=null;
BufferedOutputStream out =null;

public static final int SERVERPORT = 6000;

String outputFile = "/sdcard/Movies/asddcopy.mp4";

@Override
public void onCreate(Bundle savedInstanceState) {

    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_test_server);

    this.serverThread = new Thread(new ServerThread());
    this.serverThread.start();

}

@Override
protected void onStop() {
    super.onStop();
    try {
        serverSocket.close();
    } catch (Exception e) {
        Log.d("SERVER", "Inside onStop()");
        Log.d("SERVER", Log.getStackTraceString(e));
    }
}

class ServerThread implements Runnable {

    @SuppressLint("NewApi")
    public void run() {

        try {
            serverSocket = new DatagramSocket(SERVERPORT);
            incomingData = new byte[BUF_SIZE];
            //outgoingData = new byte[512];
            outgoingData = msg.getBytes();

            long startRxPackets = TrafficStats.getUidRxPackets(Process.myUid());
            long startTime = System.nanoTime();

            out = new BufferedOutputStream(new FileOutputStream(outputFile, true));


            while (!Thread.currentThread().isInterrupted()) {

                //serverSocket.setSoTimeout(5000);
                while (true) {
                    try{
                        //DatagramPacket incomingPacket = new DatagramPacket(incomingData, incomingData.length);
                        DatagramPacket incomingPacket = new DatagramPacket(incomingData, BUF_SIZE);

                        serverSocket.receive(incomingPacket);
                        byte[] data = incomingPacket.getData();
                        //out.write(data,0,incomingPacket.getLength());
                        //String msg = new String(incomingPacket.getData());

                        ByteArrayInputStream in = new ByteArrayInputStream(data);
                        ObjectInputStream is = new ObjectInputStream(in);

                        if (is == null) {
                            is = new ObjectInputStream(in);
                        }
                        Message msg = (Message) is.readObject();
                        System.out.println(msg.getSeqNo());
                        /*if ("END".equals(msg.substring(0, 3).trim())) {
                            Log.d("SERVER", "Inside END condition");
                            break;
                        }*/
                        out.write(msg.getData(),0,msg.getData().length);
                        numPackRecv += 1;

                        Log.d("SERVER", "Packet Received: " + numPackRecv);

                        InetAddress client = incomingPacket.getAddress();
                        int client_port = incomingPacket.getPort();
                        DatagramPacket outgoingPacket = new DatagramPacket(outgoingData, outgoingData.length, client, client_port);
                        serverSocket.send(outgoingPacket);
                        ackSent += 1;

                        //Log.d("SERVER","Packet Received: " + numPackRecv + " :: " + "Ack Sent: " + ackSent);
                    }catch(Exception e) {
                        Log.d("SERVER", "Inside run() ex1");
                        Log.d("SERVER", Log.getStackTraceString(e));
                        break;

                    }
                }

                out.close();

                serverSocket.disconnect();
                serverSocket.close();

                Log.d("SERVER", "Transfer Complete");
                 Log.d("SERVER", "Actual Time elapsed = " + (System.nanoTime() - startTime)/Math.pow(10, 9) + " s");
                 Log.d("SERVER", "Total Packets Received = " + Long.toString(TrafficStats.getUidRxPackets(Process.myUid()) - startRxPackets));
                 Log.d("SERVER", "Packets Received from Socket = " + numPackRecv);
                 break;
            }
            out.close();
            serverSocket.disconnect();
            serverSocket.close();

            /* Log.d("SERVER", "Transfer Complete");
             Log.d("SERVER", "Actual Time elapsed = " + (System.nanoTime() - startTime)/Math.pow(10, 9) + " s");
             Log.d("SERVER", "Total Packets Received = " + Long.toString(TrafficStats.getUidRxPackets(Process.myUid()) - startRxPackets));
             Log.d("SERVER", "Packets Received from Socket = " + numPackRecv);*/

        }catch (Exception e) {
            Log.d("SERVER", "Inside run() ex2");
            Log.d("SERVER", Log.getStackTraceString(e));
            serverSocket.disconnect();
            serverSocket.close();
            }
        }
    }

这是客户端

    public class TestClient extends Activity { private DatagramSocket clientSocket;
    byte[] incomingData;
    int BUF_SIZE = 500;
    int numBytesRead = 0;
    int numPackSent = 0;

    private static final int SERVERPORT = 6000;
    private static final String SERVER_IP = "10.0.0.22";

    String inFile = "/sdcard/Movies/asdd.mp4";

    @Override
    public void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_test_client);      

new Thread(new ClientThread()).start();
    }

    public void onClick(View view) {

new workInProgress().execute("");
    }

    private class workInProgress extends AsyncTask<Object, Object, Object> {

@SuppressLint("NewApi")
@Override
protected Object doInBackground(Object... params) {

    try {

        Log.d("CLIENT", "Sending a file to the server...");
        BufferedInputStream inputBuf = new BufferedInputStream(new FileInputStream(inFile));
        //byte[] fileBytes = new byte[(int) inFile.length()];
        byte[] fileBytes = new byte[BUF_SIZE];

        incomingData = new byte[BUF_SIZE];
        double numPktToSend = Math.ceil(inFile.length()*1.0/BUF_SIZE);
        //Log.d("CLIENT", "Total packets to be sent = " + numPktToSend);

        int sleepCycle = 1;
        long sysPackSent = 0;
        //long startTxPackets = TrafficStats.getTotalTxPackets();
        long startTxPackets = TrafficStats.getUidTxPackets(Process.myUid());
        Log.d("CLIENT", "startTxPacks: " + startTxPackets);
        long packDrops = 0;
        long startTime = System.nanoTime();
        long count=0;
        long ackRec=0;
        int seqNo = 0;

        ByteArrayOutputStream outStream = new ByteArrayOutputStream();
        ObjectOutputStream os = new ObjectOutputStream(outStream);


        while((numBytesRead = inputBuf.read(fileBytes)) != -1) {
            //DatagramPacket packet = new DatagramPacket(fileBytes, fileBytes.length);
             if (os == null) {
                    os = new ObjectOutputStream(outStream);
             }
             Message msg = new Message(++seqNo, fileBytes, false);   
             os.writeObject(msg);
             os.flush();
             os.reset();

             byte[] data = outStream.toByteArray();
             DatagramPacket packet = new DatagramPacket(data, data.length);
            clientSocket.send(packet);
            numPackSent += 1;

            //Log.d("CLIENT", "No of packets sent = " + numPackSent);

            sysPackSent = TrafficStats.getUidTxPackets(Process.myUid()) - startTxPackets;

            try{
                clientSocket.setSoTimeout(5000);
                packet = new DatagramPacket(incomingData, incomingData.length);
                clientSocket.receive(packet);
                String recAck = new String(packet.getData());
                ackRec++;
            }
            catch(Exception e) {
                //Log.d("CLIENT", Log.getStackTraceString(e));
            }

            packDrops = numPackSent - ackRec;
            if (packDrops > count) {
                sleepCycle = Math.min(16, sleepCycle * 2);
                count = packDrops;
                Log.d("CLIENT",String.valueOf(sleepCycle) + " :: " + numPackSent);
            } else {
                sleepCycle = Math.max(sleepCycle - 1, 1);
            }
            Thread.sleep(sleepCycle);

        }

        if (numBytesRead == -1) {
            fileBytes = "END".getBytes();
            Log.d("CLIENT", "Sending END Packet");
            clientSocket.send(new DatagramPacket(fileBytes, fileBytes.length));
        }


        Log.d("CLIENT", "Actual Time elapsed = " + (System.nanoTime() - startTime)/Math.pow(10, 9) + " s");
        Log.d("CLIENT", "Total Packets Transmitted = " + Long.toString(sysPackSent));
        Log.d("CLIENT", "No of packets dropped = " + String.valueOf(packDrops));
        Log.d("CLIENT", "Packets Pushed to Socket = " + numPackSent);
        Log.d("CLIENT", "Number of Acknoledgments received " +ackRec);

        inputBuf.close();
        os.close();
        outStream.close();
        clientSocket.disconnect();
        clientSocket.close();

        Log.d("CLIENT", "Sending file.. Complete!!!");



} catch (Exception e) {
    Log.d("CLIENT", Log.getStackTraceString(e));
    clientSocket.disconnect();
    clientSocket.close();
}

    return null;
}

    }

    class ClientThread implements Runnable  {

@Override
public void run() {

    try {
        InetAddress serverAddr = InetAddress.getByName(SERVER_IP);
        clientSocket = new DatagramSocket();
        clientSocket.connect(serverAddr, SERVERPORT);
        Log.d("CLIENT", "Connection Successful");
    } catch (UnknownHostException e1) {
        Log.d("CLIENT", "Inside run() UnknownHostEx");
        Log.d("CLIENT", Log.getStackTraceString(e1));
    } catch (IOException e1) {
        Log.d("CLIENT", "Inside run() IOEx");
        Log.d("CLIENT", Log.getStackTraceString(e1));
    }

}

    }

我在服务器端遇到一些错误:

1. 我收到的每个数据包的序列号相同(即 1)
2. 我不确定传入数据包的缓冲区大小，因为我在客户端使用 500 字节，在服务器端使用 1024 字节。如果我在两个代码中都使用 500 个字节，则会出现文件结束异常。

如果您能提出更好的方法来实现同样的事情，我将非常感激!谢谢:)

谢谢!

Answer 1

1. 创建一个ByteArrayOutputStream。
2. 将其包装在 DataOutputStream
中
3. 使用DataOutputStream.writeInt()写入序列号。
4. 使用write()写入数据。
5. 根据 ByteArrayOutputStream 返回的字节数组构造 DatagramPacket。

在接收方，执行相反的操作，在每种情况下使用互补的类和方法。这些内容留给读者作为练习。