尝试使用非阻塞IO打包TCP很难!难道我做错了什么？

哦,我希望TCP是基于数据包的,就像UDP一样![见评论]但是,唉,情况并非如此,所以我正在尝试实现自己的数据包层.这是迄今为止的事件链(忽略写入数据包)

哦,我的数据包非常简单:长度为两个无符号字节,然后是byte [length]数据.(我无法想象如果它们更复杂,我会在if声明中听到我的声音!)

Server处于无限循环中,接受连接并将它们添加到Connections 列表中.

PacketGatherer(另一个线程)使用a Selector来确定哪些Connection.SocketChannels已准备好读取.

它遍历结果并告诉每个Connection人read().

每个Connection都有一个部分IncomingPacket和一个Packet已完全读取并等待处理的s 列表.

上read():

告诉部分IncomingPacket人阅读更多数据.(IncomingPacket.readData下)

如果它已经完成了read(IncomingPacket.complete()),则从中创建一个Packet并粘贴Packet到等待处理的列表中,然后用新的替换它IncomingPacket.

这有几个问题.首先,一次只读取一个数据包.如果IncomingPacket只需要一个字节,那么这个传递只读取一个字节.这当然可以通过循环来修复,但它开始变得复杂,我想知道是否有更好的整体方式.

其次,逻辑中IncomingPacket有点疯狂,能够读取两个字节的长度然后读取实际数据.以下是快速简便阅读的代码:

int readBytes;         // number of total bytes read so far
byte length1, length2; // each byte in an unsigned short int (see getLength())

public int getLength() { // will be inaccurate if readBytes < 2
    return (int)(length1 << 8 | length2);
}

public void readData(SocketChannel c) {
    if (readBytes < 2) { // we don't yet know the length of the actual data
        ByteBuffer lengthBuffer = ByteBuffer.allocate(2 - readBytes);
        numBytesRead = c.read(lengthBuffer);

        if(readBytes == 0) {
            if(numBytesRead >= 1)
                length1 = lengthBuffer.get();

            if(numBytesRead == 2)
                length2 = lengthBuffer.get();
        } else if(readBytes == 1) {
            if(numBytesRead == 1)
                length2 = lengthBuffer.get();
        }
        readBytes += numBytesRead;
    }

    if(readBytes >= 2) { // then we know we have the entire length variable
        // lazily-instantiate data buffers based on getLength()
        // read into data buffers, increment readBytes

        // (does not read more than the amount of this packet, so it does not
        // need to handle overflow into the next packet's data)
    }
}

public boolean complete() {
    return (readBytes > 2 && readBytes == getLength()+2);
}

基本上我需要有关我的代码和整个过程的反馈.请提出任何改进建议.如果你有关于如何更好地实现整个系统的建议,即使对我的整个系统进行检修也是可以的.也欢迎预订建议; 我喜欢书.我只是觉得事情不太对劲.

这是我想出的一般解决方案,感谢Juliano的回答:(如果您有任何问题,请随时发表评论)

public void fillWriteBuffer() {
    while(!writePackets.isEmpty() && writeBuf.remaining() >= writePackets.peek().size()) {
        Packet p = writePackets.poll();
        assert p != null;
        p.writeTo(writeBuf);
    }
}

public void fillReadPackets() {
    do {
        if(readBuf.position() < 1+2) {
            // haven't yet received the length
            break;
        }

        short packetLength = readBuf.getShort(1);

        if(readBuf.limit() >= 1+2 + packetLength) {
            // we have a complete packet!

            readBuf.flip();

            byte packetType = readBuf.get();

            packetLength = readBuf.getShort();

            byte[] packetData = new byte[packetLength];
            readBuf.get(packetData);

            Packet p = new Packet(packetType, packetData);
            readPackets.add(p);
            readBuf.compact();
        } else {
            // not a complete packet
            break;
        }

    } while(true);
}

Juliano.. 7

可能这不是您正在寻找的答案,但有人必须说出来:您可能过度设计解决方案以解决一个非常简单的问题.

在完全到达之前你没有数据包,甚至没有数据包IncomingPacket.您只有一个没有定义含义的字节流.通常,简单的解决方案是将输入数据保存在缓冲区中(它可以是一个简单的byte []数组,但如果性能有问题,建议使用适当的弹性和循环缓冲区).每次读取后,检查缓冲区的内容以查看是否可以从那里提取整个数据包.如果可以的话,构建你的Packet,从缓冲区的开头丢弃正确的字节数并重复.如果或者当您无法提取整个数据包时,请将这些传入的字节保留在那里,直到下次从套接字中成功读取内容为止.

如果您正在使用它,如果您正在通过流通道进行基于数据报的通信,我建议您在每个"数据包"的开头包含一个幻数,以便您可以测试连接的两端是否仍然是同步的.如果由于某种原因(一个错误),它们之一可能会从流中读取或写入错误的字节数,它们可能会失去同步.

可能这不是您正在寻找的答案,但有人必须说出来:您可能过度设计解决方案以解决一个非常简单的问题.

在完全到达之前你没有数据包,甚至没有数据包IncomingPacket.您只有一个没有定义含义的字节流.通常,简单的解决方案是将输入数据保存在缓冲区中(它可以是一个简单的byte []数组,但如果性能有问题,建议使用适当的弹性和循环缓冲区).每次读取后,检查缓冲区的内容以查看是否可以从那里提取整个数据包.如果可以的话,构建你的Packet,从缓冲区的开头丢弃正确的字节数并重复.如果或者当您无法提取整个数据包时,请将这些传入的字节保留在那里,直到下次从套接字中成功读取内容为止.

如果您正在使用它,如果您正在通过流通道进行基于数据报的通信,我建议您在每个"数据包"的开头包含一个幻数,以便您可以测试连接的两端是否仍然是同步的.如果由于某种原因(一个错误),它们之一可能会从流中读取或写入错误的字节数,它们可能会失去同步.