深刻淺出NIO之Channel、Buffer

時間 2019-11-08

標籤深刻 nio channel buffer 欄目 Netty 简体版

原文原文鏈接

前言

Java NIO 由如下幾個核心部分組成：
1 、Buffer
二、Channel
三、Selector數組

傳統的IO操做面向數據流，意味着每次從流中讀一個或多個字節，直至完成，數據沒有被緩存在任何地方。緩存

NIO操做面向緩衝區，數據從Channel讀取到Buffer緩衝區，隨後在Buffer中處理數據。app

本文着重介紹Channel和Buffer的概念以及在文件讀寫方面的應用和內部實現原理。dom

Buffer

A buffer is a linear, finite sequence of elements of a specific primitive type.異步

一塊緩存區，內部使用字節數組存儲數據，並維護幾個特殊變量，實現數據的反覆利用。
一、mark：初始值爲-1，用於備份當前的position;
二、position：初始值爲0，position表示當前能夠寫入或讀取數據的位置，當寫入或讀取一個數據後，position向前移動到下一個位置；
三、limit：寫模式下，limit表示最多能往Buffer裏寫多少數據，等於capacity值；讀模式下，limit表示最多能夠讀取多少數據。
四、capacity：緩存數組大小jvm

mark()：把當前的position賦值給marksocket

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

reset()：把mark值還原給positionthis

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

clear()：一旦讀完Buffer中的數據，須要讓Buffer準備好再次被寫入，clear會恢復狀態值，但不會擦除數據。spa

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

flip()：Buffer有兩種模式，寫模式和讀模式，flip後Buffer從寫模式變成讀模式。code

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

rewind()：重置position爲0，從頭讀寫數據。

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

目前Buffer的實現類有如下幾種：

ByteBuffer
CharBuffer
DoubleBuffer
FloatBuffer
IntBuffer
LongBuffer
ShortBuffer
MappedByteBuffer

ByteBuffer

A byte buffer，extend from Buffer

ByteBuffer的實現類包括"HeapByteBuffer"和"DirectByteBuffer"兩種。

HeapByteBuffer

public static ByteBuffer allocate(int capacity) {
    if (capacity < 0)
        throw new IllegalArgumentException();
    return new HeapByteBuffer(capacity, capacity);
}
HeapByteBuffer(int cap, int lim) {  
    super(-1, 0, lim, cap, new byte[cap], 0);
}

HeapByteBuffer經過初始化字節數組hd，在虛擬機堆上申請內存空間。

DirectByteBuffer

public static ByteBuffer allocateDirect(int capacity) {
    return new DirectByteBuffer(capacity);
}
DirectByteBuffer(int cap) {
    super(-1, 0, cap, cap);
    boolean pa = VM.isDirectMemoryPageAligned();
    int ps = Bits.pageSize();
    long size = Math.max(1L, (long)cap + (pa ? ps : 0));
    Bits.reserveMemory(size, cap);

    long base = 0;
    try {
        base = unsafe.allocateMemory(size);
    } catch (OutOfMemoryError x) {
        Bits.unreserveMemory(size, cap);
        throw x;
    }
    unsafe.setMemory(base, size, (byte) 0);
    if (pa && (base % ps != 0)) {
        // Round up to page boundary
        address = base + ps - (base & (ps - 1));
    } else {
        address = base;
    }
    cleaner = Cleaner.create(this, new Deallocator(base, size, cap));
    att = null;
}

DirectByteBuffer經過unsafe.allocateMemory申請堆外內存，並在ByteBuffer的address變量中維護指向該內存的地址。
unsafe.setMemory(base, size, (byte) 0)方法把新申請的內存數據清零。

Channel

A channel represents an open connection to an entity such as a hardware device, a file, a network socket, or a program component that is capable of performing one or more distinct I/O operations, for example reading or writing.

NIO把它支持的I/O對象抽象爲Channel，Channel又稱「通道」，相似於原I/O中的流（Stream），但有所區別：
一、流是單向的，通道是雙向的，可讀可寫。
二、流讀寫是阻塞的，通道能夠異步讀寫。
三、流中的數據能夠選擇性的先讀到緩存中，通道的數據老是要先讀到一個緩存中，或從緩存中寫入，以下所示：

目前已知Channel的實現類有：

FileChannel
DatagramChannel
SocketChannel
ServerSocketChannel

FileChannel

A channel for reading, writing, mapping, and manipulating a file.
一個用來寫、讀、映射和操做文件的通道。

FileChannel的read、write和map經過其實現類FileChannelImpl實現。

read實現

public int read(ByteBuffer dst) throws IOException {
    ensureOpen();
    if (!readable)
        throw new NonReadableChannelException();
    synchronized (positionLock) {
        int n = 0;
        int ti = -1;
        try {
            begin();
            ti = threads.add();
            if (!isOpen())
                return 0;
            do {
                n = IOUtil.read(fd, dst, -1, nd);
            } while ((n == IOStatus.INTERRUPTED) && isOpen());
            return IOStatus.normalize(n);
        } finally {
            threads.remove(ti);
            end(n > 0);
            assert IOStatus.check(n);
        }
    }
}

FileChannelImpl的read方法經過IOUtil的read實現：

static int read(FileDescriptor fd, ByteBuffer dst, long position,
                NativeDispatcher nd) IOException {
    if (dst.isReadOnly())
        throw new IllegalArgumentException("Read-only buffer");
    if (dst instanceof DirectBuffer)
        return readIntoNativeBuffer(fd, dst, position, nd);

    // Substitute a native buffer
    ByteBuffer bb = Util.getTemporaryDirectBuffer(dst.remaining());
    try {
        int n = readIntoNativeBuffer(fd, bb, position, nd);
        bb.flip();
        if (n > 0)
            dst.put(bb);
        return n;
    } finally {
        Util.offerFirstTemporaryDirectBuffer(bb);
    }
}

經過上述實現能夠看出，基於channel的文件數據讀取步驟以下：
一、申請一塊和緩存同大小的DirectByteBuffer bb。
二、讀取數據到緩存bb，底層由NativeDispatcher的read實現。
三、把bb的數據讀取到dst（用戶定義的緩存，在jvm中分配內存）。
read方法致使數據複製了兩次。

write實現

public int write(ByteBuffer src) throws IOException {
    ensureOpen();
    if (!writable)
        throw new NonWritableChannelException();
    synchronized (positionLock) {
        int n = 0;
        int ti = -1;
        try {
            begin();
            ti = threads.add();
            if (!isOpen())
                return 0;
            do {
                n = IOUtil.write(fd, src, -1, nd);
            } while ((n == IOStatus.INTERRUPTED) && isOpen());
            return IOStatus.normalize(n);
        } finally {
            threads.remove(ti);
            end(n > 0);
            assert IOStatus.check(n);
        }
    }
}

和read實現同樣，FileChannelImpl的write方法經過IOUtil的write實現：

static int write(FileDescriptor fd, ByteBuffer src, long position,
                 NativeDispatcher nd) throws IOException {
    if (src instanceof DirectBuffer)
        return writeFromNativeBuffer(fd, src, position, nd);
    // Substitute a native buffer
    int pos = src.position();
    int lim = src.limit();
    assert (pos <= lim);
    int rem = (pos <= lim ? lim - pos : 0);
    ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);
    try {
        bb.put(src);
        bb.flip();
        // Do not update src until we see how many bytes were written
        src.position(pos);
        int n = writeFromNativeBuffer(fd, bb, position, nd);
        if (n > 0) {
            // now update src
            src.position(pos + n);
        }
        return n;
    } finally {
        Util.offerFirstTemporaryDirectBuffer(bb);
    }
}

經過上述實現能夠看出，基於channel的文件數據寫入步驟以下：
一、申請一塊DirectByteBuffer，bb大小爲byteBuffer中的limit - position。
二、複製byteBuffer中的數據到bb中。
三、把數據從bb中寫入到文件，底層由NativeDispatcher的write實現，具體以下：

private static int writeFromNativeBuffer(FileDescriptor fd, 
        ByteBuffer bb, long position, NativeDispatcher nd)
    throws IOException {
    int pos = bb.position();
    int lim = bb.limit();
    assert (pos <= lim);
    int rem = (pos <= lim ? lim - pos : 0);

    int written = 0;
    if (rem == 0)
        return 0;
    if (position != -1) {
        written = nd.pwrite(fd,
                            ((DirectBuffer)bb).address() + pos,
                            rem, position);
    } else {
        written = nd.write(fd, ((DirectBuffer)bb).address() + pos, rem);
    }
    if (written > 0)
        bb.position(pos + written);
    return written;
}

write方法也致使了數據複製了兩次

Channel和Buffer示例

File file = new RandomAccessFile("data.txt", "rw");
FileChannel channel = file.getChannel();
ByteBuffer buffer = ByteBuffer.allocate(48);

int bytesRead = channel.read(buffer);
while (bytesRead != -1) {
    System.out.println("Read " + bytesRead);
    buffer.flip();
    while(buffer.hasRemaining()){
        System.out.print((char) buffer.get());
    }
    buffer.clear();
    bytesRead = channel.read(buffer);
}
file.close();

注意buffer.flip() 的調用，首先將數據寫入到buffer，而後變成讀模式，再從buffer中讀取數據。