通过非阻塞 Java NIO 套接字传递对象并不总是成功

Question

我正在编写一个程序来使用 Java NIO 非阻塞套接字模拟 P2P 网络上的对等点。这个想法是让每个对等点使用相同的代码来发送和接收消息以及允许对等点引导到网络的服务器。

我遇到的问题是，尽管最多四个对等点可以成功加入网络并相互通信（Ping、Pong、Query 和 QueryHit），但当我添加第五个对等点时，服务器总是报告“StreamCorruptedException”。我已经检查过这个网站以及Java NIO代码/教程网站以获取解决方案，但无济于事。我知道通过非阻塞套接字发送对象并不容易/理想（特别是使用 ObjectOutputStream 和 ObjectInputStream），但我想尽量减少线程的使用（而且我不想从头开始重写！）。

我将首先展示最重要的方法（消息发送和接收），但如果需要，我可以稍后添加更多方法。

写入方法：

public void write(SelectionKey selKey){ 
    SocketChannel channel = (SocketChannel)selKey.channel();
    ArrayList<Message> queue = pending.get(channel);

    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    ObjectOutputStream oos;
    try{
        oos = new ObjectOutputStream(baos);
    }catch(Exception e){
        System.err.println("Could not create object output stream. Aborting...");
        return;
    }

    while(!queue.isEmpty()){            
        Message message = queue.get(0);
        buffer.clear();
        try{
            oos.writeObject(message);
            buffer = ByteBuffer.wrap(baos.toByteArray());
            channel.write(buffer);
            oos.flush();
            baos.flush();
        }catch(Exception e){
            System.err.println("Could not parse object. Aborting...");
            queue.remove(0);
            return;
        }
        queue.remove(0);
    }
    selKey.interestOps(SelectionKey.OP_READ);
}

和读取方法：

public Message read(SelectionKey selKey) throws IOException, ClassNotFoundException{
    SocketChannel channel = (SocketChannel)selKey.channel();        
    Message message = null;

    buffer = ByteBuffer.allocate(8192);

    int bytesRead = channel.read(buffer);
    if(bytesRead > 0){
        buffer.flip();
        InputStream bais = new ByteArrayInputStream(buffer.array(), 0, buffer.limit());
        ObjectInputStream ois = new ObjectInputStream(bais); //Offending line. Produces the StreamCorruptedException.
        message = (Message)ois.readObject();
        ois.close();
    }

    return message;
}

任何帮助将不胜感激！

Answer 1

int bytesRead = channel.read(buffer);
if(bytesRead > 0){
    buffer.flip();
    InputStream bais = new ByteArrayInputStream(buffer.array(), 0, buffer.limit());
    ObjectInputStream ois = new ObjectInputStream(bais); //Offending line. Produces the StreamCorruptedException.
    message = (Message)ois.readObject();
    ois.close();
}

在这里，您混合了阻塞和非阻塞 I/O。缓冲区不会从套接字中彻底读取，它只会读取可用的内容。如果您打算采用这种方法，您需要首先将所有数据读入缓冲区。

Answer 2

您不能假设您的读取将返回一个块中的整个对象。您需要一个“带外”（好吧，在 ObjectStream 带外）协议来确保您接收完整的消息，并根据需要从读取的块中组装消息。对于非阻塞来说，这将比现在复杂得多。