Java Socket InputStream 读取丢失的字节

Question

通过从套接字的输入流读取字节，我得到了一个非常奇怪的行为。

在我的项目中，客户确实向服务发出请求。对于每个请求，都会建立一个新连接。

首先发送字节，告诉服务将遵循哪种请求。

然后请求本身被发送。

服务接收字节并处理请求。这确实可以满足至少 95% 的请求。对于剩下的 5%，有一种我无法弄清楚的奇怪行为。

这些字节并不是发送的所有字节。但这个主题最奇怪的事情是丢失的字节不在流的开头或结尾。它们分布在整个流中。

遗憾的是，我无法在这里提供完整的代码，因为它与工作相关。但我可以提供显示问题本身的测试代码。

为了弄清楚发生了什么，我写了 2 个类。一种源自 java.net.Socket，另一种源自 java.net.ServerSocket。

代码如下:

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;



public class DebugSocket extends Socket
{
    private class InputStreamWrapper extends InputStream
    {
        private int
            availables,
            closes,
            marksupporteds,
            resets;

        private List<Integer>
            marks   = new ArrayList<Integer>(),
            reads   = new ArrayList<Integer>();

        private List<Long>
            skips   = new ArrayList<Long>();


        @Override
        public int available() throws IOException
        {
            availables++;
            return DebugSocket.this.origininput.available();
        }

        @Override
        public void close() throws IOException
        {
            closes++;
            DebugSocket.this.origininput.close();
        }

        @Override
        public synchronized void mark(int readlimit)
        {
            marks.add(readlimit);
            DebugSocket.this.origininput.mark(readlimit);
        }

        @Override
        public boolean markSupported()
        {
            marksupporteds++;
            return DebugSocket.this.origininput.markSupported();
        }

        @Override
        public synchronized void reset() throws IOException
        {
            resets++;
            DebugSocket.this.origininput.reset();
        }

        @Override
        public int read() throws IOException
        {
            int read = DebugSocket.this.origininput.read();

            reads.add(read);

            if ( read != -1 )
            {
                DebugSocket.this.inputdebugbuffer.write(read);
            }

            return read;
        }

        @Override
        public int read(byte[] b) throws IOException
        {
            int read = DebugSocket.this.origininput.read(b);

            DebugSocket.this.inputdebugbuffer.write(b, 0, read);

            return read;
        }

        @Override
        public int read(byte[] b, int off, int len) throws IOException
        {
            int read = DebugSocket.this.origininput.read(b, off, len);

            DebugSocket.this.inputdebugbuffer.write(b, off, read);

            return read;
        }

        @Override
        public long skip(long n) throws IOException
        {
            long skipped = DebugSocket.this.origininput.skip(n);

            skips.add(skipped);

            return skipped;
        }
    }

    private class OutputStreamWrapper extends OutputStream
    {
        private int
            flushes,
            closes;


        @Override
        public void close() throws IOException
        {
            closes++;
            DebugSocket.this.originoutput.close();
        }

        @Override
        public void flush() throws IOException
        {
            flushes++;
            DebugSocket.this.originoutput.flush();
        }

        @Override
        public void write(int b) throws IOException
        {
            DebugSocket.this.outputdebugbuffer.write(b);
            DebugSocket.this.originoutput.write(b);
            DebugSocket.this.originoutput.flush();
        }

        @Override
        public void write(byte[] b) throws IOException
        {
            DebugSocket.this.outputdebugbuffer.write(b);
            DebugSocket.this.originoutput.write(b);
            DebugSocket.this.originoutput.flush();
        }

        @Override
        public void write(byte[] b, int off, int len) throws IOException
        {
            DebugSocket.this.outputdebugbuffer.write(b, off, len);
            DebugSocket.this.originoutput.write(b, off, len);
            DebugSocket.this.originoutput.flush();
        }
    }


    private static final Object
        staticsynch = new Object();

    private static long
        idcounter   = 0;


    private final long
        id;

    private final ByteArrayOutputStream
        inputdebugbuffer,
        outputdebugbuffer;

    private final InputStream
        inputwrapper;

    private final OutputStream
        outputwrapper;

    private InputStream
        origininput;

    private OutputStream
        originoutput;


    public InputStream getInputStream() throws IOException
    {
        if ( origininput == null )
        {
            synchronized ( inputdebugbuffer )
            {
                if ( origininput == null )
                {
                    origininput = super.getInputStream();
                }
            }
        }

        return inputwrapper;
    }

    public OutputStream getOutputStream() throws IOException
    {
        if ( originoutput == null )
        {
            synchronized ( outputdebugbuffer )
            {
                if ( originoutput == null )
                {
                    originoutput    = super.getOutputStream();
                }
            }
        }

        return outputwrapper;
    }


    public DebugSocket()
    {
        id                  = getNextId();
        inputwrapper        = new InputStreamWrapper();
        outputwrapper       = new OutputStreamWrapper();
        inputdebugbuffer    = new ByteArrayOutputStream();
        outputdebugbuffer   = new ByteArrayOutputStream();
    }


    private static long getNextId()
    {
        synchronized ( staticsynch )
        {
            return ++idcounter;
        }
    }
}

import java.io.IOException;
import java.net.ServerSocket;


public class DebugServerSocket extends ServerSocket
{
    public DebugServerSocket() throws IOException
    {
        super();
    }

    public DebugSocket accept() throws IOException
    {
        DebugSocket s = new DebugSocket();

        implAccept(s);

        return s;
    }
}

类 DebugSocket 获取与 InputStream 以及 OutputStream 的每次交互的通知

现在，当问题发生时，我总是可以看到字节丢失。

这里是一个例子:

客户端发送1758字节。我从 DebugSocket 中的成员 outputdebugbuffer 获取了前 23 个字节。

Bytes: 0,0,0,0,0,0,0,2,0,0,6,-46,31,-117,8,0,0,0,0,0,0,0,-83

服务器收到 227 字节。对于调试问题，我总是读取输入流，直到得到 -1，以便所有字节都继续进行。现在，我从 DebugSocket 中的成员 inputdebugbuffer 获得了服务器端的 16 个前导字节。

Bytes: 0,0,0,6,-46,31,-117,8,0,0,0,0,0,0,0,-83

如图所示，缺少 7 个字节。前 8 个字节是一个长值，我将其更改为字节值以进行调试。所以我认为第一个字节总是正确的。

如果代码出现故障，则不会继续处理任何请求，但正如我之前所说，这种情况最多只发生在所有连接的 5% 上。

有人知道这里发生了什么吗？

我还使用了DataInputStream和DataOutputStream来发送数据。我总是在每次写入操作后刷新，正如您在 DebugSocket 的 OutputStreamWrapper 中看到的那样。

我错过了什么吗？

如果需要其他代码，我会尝试发布它。

附注该服务是多线程的，并行处理 100 个请求。此外，客户端是多线程的，可以并行执行 20 个请求。如前所述，每个请求都使用其一个连接，并在请求继续后立即关闭该连接。

我希望有人对这个问题有想法。

编辑:

没有主要方法来显示它执行评论中所要求的任何操作，但这里使用的是客户端和服务器的代码块。

客户端:(在 20 个线程中并行运行)

    public void sendRequest(long _requesttype, byte[] _bytes)
    {
        Socket              socket  = null;
        DataInputStream     input   = null;
        DataOutputStream    output  = null;
        InputStream         sinput  = null;
        OutputStream        soutput = null;

        try
        {
            socket  = new DebugSocket();

            socket.connect(serveraddress);

            sinput  = socket.getInputStream();
            soutput = socket.getOutputStream();

            input   = new DataInputStream(sinput);
            output  = new DataOutputStream(soutput);

            output.writeLong(_requesttype);

            output.flush();
            soutput.flush();

            output.write(_bytes);

            output.flush();
            soutput.flush();

            // wait for notification byte that service had received all data.
            input.readByte();
        }
        catch (IOException ex)
        {
            LogHelper.log(ex);
        }
        catch (Error err)
        {
            throw err;
        }
        finally
        {
            output.flush();
            soutput.flush();

            input.close();
            output.close();

            finishSocket(socket);
        }
    }

服务器:(每个请求在一个线程中运行。最多 100 个线程)

    public void proceedRequest(DebugSocket _socket)
    {
        DataInputStream     input   = null;
        DataOutputStream    output  = null;
        InputStream         sinput  = null;
        OutputStream        soutput = null;

        try
        {
            sinput  = _socket.getInputStream();
            soutput = _socket.getOutputStream();

            input   = new DataInputStream(sinput);
            output  = new DataOutputStream(soutput);

            RequestHelper.proceed(input.readLong(), input, output);

            // send notification byte to the client.
            output.writeByte(1);

            output.flush();
            soutput.flush();
        }
        catch (IOException ex)
        {
            LogHelper.log(ex);
        }
        catch (Error err)
        {
            throw err;
        }
        finally
        {
            output.flush();
            soutput.flush();

            input.close();
            output.close();
        }
    }

在服务器代码中，readLong() 已经因丢失字节而失败。

Answer 1

好的，我已经尝试了所有可能的方法来查找原因。根据我在套接字编程和并行处理方面的经验，我可以说代码本身没有错误。嗅探器也告诉我这一点。我的机器上的某些东西干扰了传输。

我停用了所有我能想到的(防火墙/防病毒/恶意软件扫描仪)，但没有效果。

有人知道还有什么可能会扰乱 tcp 包吗？

编辑:

好的，我明白了。 AVG 2014 很困惑。 Jetzt 停用组件不起作用。在“选项”->“设置”中有一个菜单点，您可以在其中停用 AVG 保护。

有人了解这个主题吗？