Java NIO、Channel、Selector 详解

// 恒等式: mark <= position <= limit <= capacity
private int mark = -1;
private int position = 0;
private int limit;
private int capacity;

// 仅在 direct buffers 中使用
long address;

// 该字节数组仅在分配在堆上时才非空（参考下面的 Direct vs. non-direct buffers）
final byte[] hb;

public static ByteBuffer allocate(int capacity) {
    if (capacity < 0)
        throw new IllegalArgumentException();
    return new HeapByteBuffer(capacity, capacity);
}

ByteBuffer byteBuf = ByteBuffer.allocate(1024);

public static ByteBuffer wrap(byte[] array) {
    ...
}

// 填充一个 byte
public abstract ByteBuffer put(byte b);

// 在指定位置填充一个 byte
public abstract ByteBuffer put(int index, byte b);

// 批量将 src buffer 填充到本 buffer
public ByteBuffer put(ByteBuffer src) {
    ...
}

// 批量将 src 数组的特定区间填充到本 buffer
public ByteBuffer put(byte[] src, int offset, int length) {
    ...
}

// 批量将 src 数组填充到本 buffer
public final ByteBuffer put(byte[] src) {
    ...
}

int read = channel.read(buffer);

public final Buffer flip() {
    limit = position; // 将 limit 设置为实际写入的数据数量
    position = 0; // 重置 position
    mark = -1; // 将 mark 设置为未标记
    return this;
}

// 读取当前 position 的字节，然后 position 自增 1
public abstract byte get();

// 读取 index 的字节（ position 不会自增！）
public abstract byte get(int index);

// 批量将缓冲区数据传递到 dst 数组中，position 自增 dst 的长度
public ByteBuffer get(byte[] dst) {
    ...
}

// 批量将缓冲区数据传递到 dst 数组中
public ByteBuffer get(byte[] dst, int offset, int length) {
    ...
}

int write = channel.write(buffer);

public final Buffer mark() {
    mark = position;
    return this;
}

public final Buffer reset() {
    int m = mark;
    if (m < 0)
        throw new InvalidMarkException();
    position = m;
    return this;
}

public final Buffer rewind() {
    position = 0;
    mark = -1;
    return this;
}

public final Buffer clear() {
    position = 0;
    limit = capacity;
    mark = -1;
    return this;
}

public abstract ByteBuffer compact();

0 <= mark <= position <= limit <= capacity

public interface Channel extends Closeable {

    public boolean isOpen();

    public void close() throws IOException;
}

int read = channel.read(buffer);

int write = channel.write(buffer);

@Test
public void testFileChannelRead() throws IOException {
    // 获取文件的 FileChannel
    FileInputStream fileInputStream = new FileInputStream("/abc");
    FileChannel channel = fileInputStream.getChannel();

    // 创建 ByteBuffer
    ByteBuffer buffer = ByteBuffer.allocate(30);

    // 将文件内容读取到 buffer 中
    channel.read(buffer);

    // buffer 从写模式，切换到读模式
    buffer.flip();

    // 打印 buffer（文件）的内容
    while (buffer.hasRemaining()) {
        System.out.print((char)buffer.get());
    }
}

@Test
public void testFileChannelWrite() throws IOException {
    // 获取文件的 FileChannel
    FileOutputStream fileOutputStream = new FileOutputStream("/abc");
    FileChannel channel = fileOutputStream.getChannel();

    // 创建 ByteBuffer
    ByteBuffer buffer = ByteBuffer.allocate(30);
    buffer.put("123456".getBytes());

    // Buffer 切换为读模式
    buffer.flip();
    while(buffer.hasRemaining()) {
        // 将 Buffer 中的内容写入文件
        channel.write(buffer);
    }
}

@Test
public void testSocketChannel() throws IOException {
    SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("localhost", 80));

    // 创建 ByteBuffer
    ByteBuffer buffer = ByteBuffer.allocate(30);
    // 读取数据
    socketChannel.read(buffer);

    // 写入数据到网络连接中
    while(buffer.hasRemaining()) {
        socketChannel.write(buffer);
    }
}

// 实例化
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 监听 8080 端口
serverSocketChannel.socket().bind(new InetSocketAddress(8080));

while (true) {
    // 一旦有一个 TCP 连接进来，就对应创建一个 SocketChannel 进行处理
    SocketChannel socketChannel = serverSocketChannel.accept();
}

这里可以看到 SocketChannel 的另一种实例化方式，SocketChannel 可读可写，操作一个网络通道。

Selector selector = Selector.open();

// 将通道设置为非阻塞模式，因为默认都是阻塞模式的
channel.configureBlocking(false);
// 注册
SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

// 通道有数据可读
public static final int OP_READ = 1 << 0;
// 可以向通道中写数据
public static final int OP_WRITE = 1 << 2;
// 成功建立 TCP 连接
public static final int OP_CONNECT = 1 << 3;
// 接受 TCP 连接
public static final int OP_ACCEPT = 1 << 4;

Selector selector = Selector.open();

channel.configureBlocking(false);

SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

while(true) {
  // 判断是否有事件准备好
  int readyChannels = selector.select();
  if(readyChannels == 0) continue;

  // 遍历
  Set<SelectionKey> selectedKeys = selector.selectedKeys();
  Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
  while(keyIterator.hasNext()) {
    SelectionKey key = keyIterator.next();

    if(key.isAcceptable()) {
        // a connection was accepted by a ServerSocketChannel.

    } else if (key.isConnectable()) {
        // a connection was established with a remote server.

    } else if (key.isReadable()) {
        // a channel is ready for reading

    } else if (key.isWritable()) {
        // a channel is ready for writing
    }

    keyIterator.remove();
  }
}