javaNIO创建tcp服务器时的重要点

在使用NIO创建非阻塞tcp服务器时，几个容易出现问题的点，如下代码注释所示：

package net.yury.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author yury757
 * 非阻塞网络IO的优点在于：
 * 1、write和read操作不会阻塞线程，若网络双方同时有IO需求，则可以进行IO读写，否则读写会返回的长度是0
 * 2、在非阻塞IO中，若遇到一个很大的IO操作占用很长时间，会导致影响其他客户端的IO，非阻塞IO可以根据需要将一次大的IO根据bytebuffer大小拆分为多次小的IO，这样不影响其他客户端。所以bytebuffer在NIO中极为重要
 *      需要分多次IO是因为操作系统中也有缓冲区，不可能说想分配多大的bytebuffer，就可以一次性读取或写入这么大的bytebuffer
 */
public class NIOServer {
    public final static int BUFFER_SIZE = 128;
    private int port;
    private AtomicInteger count = new AtomicInteger(0);

    public NIOServer(int port) {
        this.port = port;
    }

    public static void main(String[] args) throws IOException {
        NIOServer nioServer = new NIOServer(8080);
        nioServer.start();
    }

    public void start() throws IOException {
        Selector selector = Selector.open();
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        ssc.bind(new InetSocketAddress("0.0.0.0", port));

        ssc.register(selector, SelectionKey.OP_ACCEPT, null);
        System.out.println("server start!");

        while (true) {
            // 阻塞selector，当有事件发生时才会唤醒
            selector.select();
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey next = iterator.next();
                // 注意：selector不会自动对已经处理完的selectionKey做删除操作，因此这里一定要移除这个selectionKey，不然下次又会对这个selectionKey做处理
                iterator.remove();
                if (next.isAcceptable()) {
                    System.out.println("accept event happen, selectionKey: " + next);
                    ServerSocketChannel channel = (ServerSocketChannel) next.channel();
                    SocketChannel sc = channel.accept();
                    System.out.println("new connection: " + sc);
                    sc.configureBlocking(false);
                    sc.register(selector, SelectionKey.OP_READ, null);
                } else if (next.isReadable()) {
                    System.out.println("read event happen, selectionKey: " + next);
                    SocketChannel channel = (SocketChannel) next.channel();
                    // 注意：无论如何一定要对selectionKey做处理，要么read，要么cancel
                    // 客户端没有调用close，程序直接退出时，会报异常java.io.IOException: 远程主机强迫关闭了一个现有的连接。
                    // 客户端正常调用close时，会触发一个读事件，并且read返回-1
                    try {
                        channelReadLengthProtocol(channel, selector, next);
                    } catch (IOException ex) {
                        ex.printStackTrace();
                        // 注意：当报错时，认为客户端异常断开了，因此要将这个selectionKey取消掉
                        next.cancel();
                    }
                } else if (next.isWritable()) {
                    System.out.println("write event happen, selectionKey: " + next);
                    SocketChannel channel = (SocketChannel)next.channel();
                    channelWrite(channel, selector, next);
                }
            }
        }
    }

    public void channelRead(SocketChannel channel, Selector selector, SelectionKey key) throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(2);
        StringBuilder sb = new StringBuilder();
        // 注意：这里需要处理半包和粘包问题
        // 半包：一个数据包被拆分到多个buffer中，读取时没有将这些buffer组合在一起或组合的方式错误导致数据异常。
        // 解决方法1：使用attachment边扩容边多次读取，或在attachment中使用bytebuffer数组，可以避免bytebuffer拷贝；解决方法2：使用自定义协议，如channelReadLengthProtocol
        // 粘包：一个buffer中包含多个数据包，多个数据包混淆在一起，无法辨别哪个是哪个
        // 下面这个代码没有解决半包问题，在遇到中文时肯定会有乱码
        while (true) {
            int read = channel.read(buffer);
            if (read == -1) {
                // 注意：-1表示客户端正常断开，因此这里要将这个selectionKey取消掉
                key.cancel();
                return;
            } else if (read == 0) {
                /**
                 * 注意：返回值为0有三种情况：
                 * 1、由于操作系统缓冲区满了等其他原因，socketChannel中读到数据
                 * 2、byteBuffer的 position = limit
                 * 3、客户端发送数据完毕
                 */
                break;
            } else {
                // 注意：需要对buffer进行读操作时，一定要flip
                buffer.flip();
                String s = StandardCharsets.UTF_8.decode(buffer).toString();
                sb.append(s);
                // 注意：要重新对buffer进行写操作时，一定要clear，或者compact
                buffer.clear();
            }
        }
        String s = sb.toString();
        System.out.println("client send: " + s);
        channel.register(selector, ~SelectionKey.OP_READ & SelectionKey.OP_WRITE, s);
    }

    public void channelWrite(SocketChannel channel, Selector selector, SelectionKey key) throws IOException {
        String s = (String)key.attachment();
        ByteBuffer buffer = StandardCharsets.UTF_8.encode("你好, count: " + count.getAndIncrement() + ", your content length is " + s.length());
        // 注意：StandardCharsets.UTF_8.encode和ByteBuffer.wrap方法返回的buffer都是已经切到了读模式，不需要在flip，再flip一次反而把所有数据都弄丢了
        while (buffer.hasRemaining()) {
            channel.write(buffer);
        }
        channel.register(selector, ~SelectionKey.OP_WRITE & SelectionKey.OP_READ);
    }

    /**
     * 这是一个基于长度协议，并使用attachment分批读取的方式处理半包问题的解决方案
     * 主要解决办法就是在传输内容前加一个4字节数字表示剩余内容体的byte长度，然后服务端分批读取直到读取到对应长度的byte即可。
     * 客户端也需要按照这种协议去发送内容，否则可能会产生问题
     * @param channel
     * @param selector
     * @param key
     * @throws IOException
     */
    public void channelReadLengthProtocol(SocketChannel channel, Selector selector, SelectionKey key) throws IOException {
        Object attachment = key.attachment();
        AttachByteBuffer attachByteBuffer;
        if (attachment == null) {
            // 获取长度
            ByteBuffer sizeBuffer = ByteBuffer.allocate(4);
            int size = 0;
            do {
                int tmp = channel.read(sizeBuffer);
                if (tmp == -1) {
                    key.cancel();
                    return;
                } else {
                    size += tmp;
                }
            } while (size != 4);
            // 获取内容的长度
            sizeBuffer.flip();
            int length = buffer2Int(sizeBuffer, 0);
            if (length <= 0) {
                System.out.println("unsupported content length: " + length);
                key.cancel();
                return;
            }

            // 初始化
            attachByteBuffer = new AttachByteBuffer(length, ByteBuffer.allocate(BUFFER_SIZE));
            key.attach(attachByteBuffer);
        }else {
            attachByteBuffer = (AttachByteBuffer)attachment;
        }

        ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);
        int read = channel.read(buffer);
        System.out.println("read size: " + read);
        if (read == -1) {
            key.cancel();
            return;
        }
        attachByteBuffer.put(buffer);

        // 如果已经读完了，则取消读事件，注册写事件
        if (attachByteBuffer.getCurrentSize() == attachByteBuffer.getRequireSize()) {
            ByteBuffer completeBuffer = attachByteBuffer.getBuffer();
            completeBuffer.flip();
            String s = StandardCharsets.UTF_8.decode(completeBuffer).toString();
            System.out.println("client send length: " + s.length());
            channel.register(selector, ~SelectionKey.OP_READ & SelectionKey.OP_WRITE, s);
        }
    }

    /**
     * 读取bytebuffer中的4个字节，转int
     * 小端字节序
     * @param buffer
     * @param start
     * @return
     */
    public int buffer2Int(ByteBuffer buffer, int start) {
        int res = 0;
        for (int i = 0; i < 4; i++) {
            res += ((buffer.get(start + i) & 0xff) << (i * 8));
        }
        return res;
    }
}

class AttachByteBuffer {
    private int requireSize;
    private int currentSize;
    private ByteBuffer buffer;

    public AttachByteBuffer(int requireSize, ByteBuffer buffer) {
        this.requireSize = requireSize;
        this.currentSize = 0;
        this.buffer = buffer;
    }

    public int getRequireSize() {
        return requireSize;
    }

    public void setRequireSize(int requireSize) {
        this.requireSize = requireSize;
    }

    public ByteBuffer getBuffer() {
        return buffer;
    }

    public void setBuffer(ByteBuffer buffer) {
        this.buffer = buffer;
    }

    public int getCurrentSize() {
        return currentSize;
    }

    public void setCurrentSize(int currentSize) {
        this.currentSize = currentSize;
    }

    public synchronized void put(ByteBuffer newBuffer) {
        newBuffer.flip();
        int size = newBuffer.limit() - newBuffer.position();
        // 如果buffer不够存储，则扩容
        if (buffer.limit() - buffer.position() < size) {
            ByteBuffer buffer2 = ByteBuffer.allocate(buffer.limit() + NIOServer.BUFFER_SIZE * 2);
            buffer.flip();
            buffer2.put(buffer);
            // 把旧的buffer替换了，旧的buffer已经没用了
            buffer = buffer2;
        }
        // 将新读取的数据写入到attachment中的buffer中
        this.currentSize += size;
        buffer.put(newBuffer);
    }
}

简易客户端如下：

package net.yury.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;

/**
 * @author yury757
 */
public class NIOClient {
    public static void main(String[] args) throws IOException {
        client2();
    }

    public static void client1() throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 8080));
        StringBuilder str = new StringBuilder();
        for (int i = 0; i < 500; i++) {
            str.append("你");
        }
        byte[] content = str.toString().getBytes(StandardCharsets.UTF_8);
        ByteBuffer buffer = ByteBuffer.wrap(content);
        while (buffer.hasRemaining()) {
            sc.write(buffer);
        }

        StringBuilder sb = new StringBuilder();
        buffer = ByteBuffer.allocate(2);
        while (sc.read(buffer) > 0) {
            buffer.flip();
            sb.append(StandardCharsets.UTF_8.decode(buffer));
            buffer.clear();
        }
        System.out.println("server response is: " + sb.toString());
        sc.close();
    }

    public static void client2() throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 8080));
        StringBuilder str = new StringBuilder();
        for (int i = 0; i < 50000; i++) {
            str.append("你");
        }
        byte[] content = str.toString().getBytes(StandardCharsets.UTF_8);
        int size = content.length;
        ByteBuffer buffer = ByteBuffer.allocate(size + 4);
        byte[] sizeByteArray = int2Byte(size);
        // 先写长度，再写内容
        buffer.put(sizeByteArray);
        buffer.put(content);

        buffer.flip();
        while (buffer.hasRemaining()) {
            int temp = sc.write(buffer);
            System.out.println("send size: " + temp);
        }

        // 接收端一样处理，省略
        StringBuilder sb = new StringBuilder();
        buffer = ByteBuffer.allocate(2);
        while (sc.read(buffer) > 0) {
            buffer.flip();
            sb.append(StandardCharsets.UTF_8.decode(buffer));
            buffer.clear();
        }
        System.out.println("server response is: " + sb.toString());
        sc.close();
    }

    /**
     * int转byte数组
     * 小段字节序
     * @param number
     * @return
     */
    public static byte[] int2Byte(int number) {
        byte[] res = new byte[4];
        for (int i = 0; i < 4; i++) {
            res[i] = (byte)(number & 0xff);
            number = number >> 8;
        }
        return res;
    }
}

Selector.select() 方法在以下情况不阻塞：

• 事件发生时：accept、read、write、客户端正常/异常关闭

• linux下，NIO的bug发生时，不阻塞；java在linux下的Selector实现有一个bug。

• 调用Selector.wakeup()时

• 调用Selector.close()时

• Selector所在的线程调用了interrupt时