java - 我的 Java 套接字代码正在损坏数据

Question

我正在编写一个程序来测试Java的网络API(旧io vs nio vs nio2)。

我有一个仅发送两个值的服务器:

1. System.nanoTime()
2. 计数器，用于计算发送的消息数。

客户端接收此数据，将远程 System.nanoTime() 与本地时间戳进行比较以计算延迟并检查计数器以确保没有数据丢失。

由于这只是一个测试，服务器和客户端运行在同一个 JVM 中。 90% 的数据传输正确率；然而，每隔一段时间，时间戳就会完全错误。看起来它可能是一个上溢/下溢错误，但我看不出它是如何引入的。以下是错误示例:

ERROR: counter 3, remoteTS -8267580102784516096, localTS 155321716184402, diff 8267735424500700498

请注意，本地时间戳 155321716184402 转换为晚上 7 点过后不久。但远程时间戳根本就是负数!如果你看一下代码，我没有做任何花哨的日期数学，它不可能是负数。我也不明白如何得到溢出错误。我认为这可能是由于大端与小端的问题，但是每个值都会出错，而不仅仅是其中的一些值。

代码(从一个稍大的测试中提取)如下:

package networkioshootout;

import static java.lang.System.out;

import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.Date;
import java.util.Random;
import java.util.concurrent.ExecutionException;


public class DebugNetwork {
  private final static int SENDCOUNT = 100;
    private final static int PORT = 9000;
    private final static int TESTLOOP = 10;
    private final static Random rn = new Random();

    public static void main(String[] args) throws IOException, InterruptedException, ExecutionException {
        long currentNanos = System.nanoTime();
        long currentMillis = System.currentTimeMillis();
        Date now = new Date();
        System.out.println(String.format("Current date/time:%s, nanos:%s, millis:%s",
                now, currentNanos, currentMillis));

        //Server
        new Server().start();

        //Client
        for(int i=0; i< TESTLOOP; i++){
            final int DATASIZE = (1+rn.nextInt(99))*8;
            clientInputstream(DATASIZE);
        }
    }

    private static void clientInputstream(int bufferSize) throws IOException, UnknownHostException {
        final byte[] internalBuffer = new byte[bufferSize+16] ;
        final ByteBuffer longExtractor = ByteBuffer.allocate(16);

        int bytesReadSoFar = 0;
        long counter = 0;

        Socket client = new Socket(InetAddress.getLocalHost(), PORT);
        InputStream in = client.getInputStream();

        byte[] data = new byte[bufferSize];
        int size = 0;

        try{
            while(-1 != (size = in.read(data))){
                for(int i=0; i < size; i++){
                    internalBuffer[i+bytesReadSoFar] = data[i];
                }
                bytesReadSoFar += size;

                if(bytesReadSoFar >= 16){
                    int values = bytesReadSoFar/16;
                    int toRead = values;
                    int remainder = bytesReadSoFar % 16;

                    for(int i=0; i< toRead; i++){
                        int j = i * 16;

                        //long remoteTS = ByteBuffer.wrap(new byte[]{internalBuffer[j+0],internalBuffer[j+1],internalBuffer[j+2],internalBuffer[j+3],internalBuffer[j+4],internalBuffer[j+5],internalBuffer[j+6],internalBuffer[j+7]}).getLong();
                        //long remoteCounter = ByteBuffer.wrap(new byte[]{internalBuffer[j+8],internalBuffer[j+9],internalBuffer[j+10],internalBuffer[j+11],internalBuffer[j+12],internalBuffer[j+13],internalBuffer[j+14],internalBuffer[j+15]}).getLong();

                        //long remoteTS = data[0] | ((int)(data[1]) << 4) | ((int)(data[1]) << 8) | ((int)(data[1]) << 12) | ((int)(data[1]) << 16) | ((int)(data[1]) << 20) | ((int)(data[1]) << 24) ;

                        longExtractor.put(internalBuffer, j, 16);
                        longExtractor.flip();
                        long remoteTS = longExtractor.getLong();
                        long remoteCounter = longExtractor.getLong();
                        longExtractor.clear();

                        if(remoteCounter != counter){
                            String error = "ERROR: Expected remote counter to be "+counter+" but it was actually "+remoteCounter;
                            //System.out.println(error);
                            throw new RuntimeException(error);
                        }
                        counter++;

                        long localTS = System.nanoTime();
                        long latency = localTS - remoteTS;
                        if(Math.abs(latency) > 1200000000) {
                            out.println(String.format("ERROR: counter %s, remoteTS %s, localTS %s, diff %s",
                                    counter, remoteTS, localTS, latency));
                            continue;
                        }


                    }

                    //System.arraycopy(data, toRead, data, 0, remainder);
                    for(int i=0; i < remainder; i++){
                        internalBuffer[i] = internalBuffer[i+toRead];
                    }
                    bytesReadSoFar = remainder;
                }
            }
        }
        finally{
            client.close();
        }
    }

    static final class Server extends Thread{

        public void run(){
            try {
                startServer();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        private static void startServer() throws IOException {
            final ServerSocket server = new ServerSocket(PORT);

            //System.out.println("Server listening on port "+PORT);

            while(true){
                final Socket c1 = server.accept();
                c1.setTcpNoDelay(true);
                //System.out.println("Client connected");
                new Thread(new Runnable() {

                    @Override
                    public void run() {
                        long totalMsgs = 0;
                        long counter = 0;
                        DataOutputStream serverout;
                        try {
                            serverout = new DataOutputStream(c1.getOutputStream());
                            for(int i=0;i<SENDCOUNT;i++){ 
                                serverout.writeLong(System.nanoTime());
                                serverout.writeLong(counter);
                                totalMsgs++;
                                counter++;
                            }
                            //System.out.println("Sent bytes to client: "+total);
                        } catch (IOException e) {
                            out.println("Messages sent:"+totalMsgs+", current counter:"+counter);
                            e.printStackTrace();
                        }
                        finally{
                            //System.out.println("Client disconnected when counter was "+counter);
                            try { c1.close(); } catch (IOException e) { e.printStackTrace();}
                        }
                    }
                }).start();
            }
        }
    }

}

编辑:由于对此有一些评论，实际的程序有客户端通过输入流、缓冲流、NIO、NIO2 连接到服务器。这是该程序的更完整(但已过时)版本: https://gist.github.com/falconair/4975243

我尚未添加数据输入流、尝试套接字选项等。我希望在进一步操作之前解决数据损坏问题。

Answer 1

该错误与您使用 data[] 和 internalBuffer[] 以及所有数据改组有关。我没有看到真正的客户端使用像这样的代码编写。任何理智的人都会使用 BufferedInputStream。

如果要测试不同缓冲区大小的效果，请使用 new DataInputStream(new BufferedInputStream(socket.getInputStream(), bufferSize)) 和 readLong() 和完全摆脱 data 和 internalBuffer 以及 longExtractor:它们只会导致不相关的问题。

以下工作完美无缺:

private static void clientInputstream(int bufferSize) throws IOException, UnknownHostException
{
    long counter = 0;

    Socket client = new Socket(InetAddress.getLocalHost(), PORT);
    DataInputStream in = new DataInputStream(new BufferedInputStream(client.getInputStream(), bufferSize));

    try
    {
        for (;;)
        {
            long remoteTS = in.readLong();
            long remoteCounter = in.readLong();
            if (remoteCounter != counter)
            {
                String error = "ERROR: Expected remote counter to be " + counter + " but it was actually " + remoteCounter;
                //System.out.println(error);
                throw new RuntimeException(error);
            }
            counter++;

            long localTS = System.nanoTime();
            long latency = localTS - remoteTS;
            if (Math.abs(latency) > 1200000000)
            {
                out.println(String.format("ERROR: counter %s, remoteTS %s, localTS %s, diff %s",
                    counter, remoteTS, localTS, latency));
                continue;
            }
        }
    }
    catch (EOFException exc)
    {
        System.out.println("EOS");
    }
    finally
    {
        client.close();
    }
}