Java NIO SocketChannels:具有多个对象和高系统负载的序列化问题

Question

我正在使用java.nio中的SocketChannels在p2p网络中的几个对等体之间发送对象.每个对等体都有一个ServerSocketChannel,其他对等体可以连接到该对等体.我通过这些SocketChannel发送序列化对象.我的代码基础基本上是来自http://rox-xmlrpc.sourceforge.net/niotut/的NIO教程

我发送的所有消息都实现了相同的接口,因此我可以在接收端进行反序列化.以下代码执行此操作(使用字节计数器更新,请参见下文):

private void readKey(SelectionKey key) throws IOException {
    SocketChannel channel = (SocketChannel) key.channel();
    ByteBuffer readBuffer = ByteBuffer.allocate(4);

    channel.read(readBuffer);
    readBuffer.rewind();
    int numBytes = readBuffer.getInt();

    readBuffer = ByteBuffer.allocate(numBytes);
    int read = channel.read(readBuffer);

    Message msg = Message.deserialize(readBuffer);  
    this.overlay.addIncomingMessage(msg);
}

通过对象的序列化完成发送,将消息序列化并添加到队列中,更改通道的兴趣操作,并发送序列化对象.

private void writeKey(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();

    synchronized (this.pendingData) {
        List<ByteBuffer> queue = (List<ByteBuffer>) this.pendingData.get(socketChannel);

        // Write until there's not more data ...
        while (!queue.isEmpty()) {
            ByteBuffer buf = (ByteBuffer) queue.get(0);

                            // UPDATE for message length
            ByteBuffer len = ByteBuffer.allocate(4);
            len.putInt(buf.remaining());
            len.rewind();

            socketChannel.write(len);
            socketChannel.write(buf);
            if (buf.remaining() > 0) {
                // ... or the socket's buffer fills up
                break;
            }
            queue.remove(0);
        }

        if (queue.isEmpty()) {
            key.interestOps(SelectionKey.OP_READ);
        }
    }       
}

只要系统负载较低,这一切都可以正常工作.当我开始发送大量消息时,不会收到写入该通道的一些消息.当我为代码添加延迟以减慢速度时,一切都按预期工作.

我认为在接收器读取通道缓冲区以创建对象之前发送多条消息时可能会出现问题,但我不知道如何解决这个问题.

我很欣赏任何提示或想法.

此致,Christoph

更新:在第一次提示之后,我添加了传输到发送端的字节数,并且只读取接收器上的那些字节数,但没有效果.

Answer 1

您似乎假设接收方将在发送方发送它们的相同块中接收消息"数据包".事实并非如此.底层套接字可以任意拆分/加入您发送的块.

为了实现这样的基于消息的协议,您需要管理消息块.您需要在读取时将套接字视为数据流,而不是假设从读取调用接收的数据缓冲区对应于单个消息.在流上实现基于消息的协议的一种方法是首先写入消息长度,然后写入消息字节.在接收端,您首先读取消息长度,然后在解析消息时只消耗那么多字节.

更新:

如果你还在丢失消息,我猜你的同步逻辑就是问题.从您包含的代码中分辨出来很难,但是如何同步各个队列(我只看到pendingData列表中的顶级锁).还有,你在接收端使用什么样的同步？