【死磕 NIO】— 深入分析Channel和FileChannel

大家好，我是大明哥，這次我們來看看NIO的第二個組件：Channel。

上篇文章[【死磕 NIO】— 深入分析Buffer]介紹了 NIO 中的 Buffer，Buffer 我們可以認(rèn)為他是裝載數(shù)據(jù)的容器，有了容器，還需要傳輸數(shù)據(jù)的通道才能完成數(shù)據(jù)的傳輸，這個通道就是今天要介紹的 Channel。

Channel 我們可以認(rèn)為它是本地 I/O 設(shè)備、網(wǎng)絡(luò) I/O 的通信橋梁，只有搭建了這座橋梁，數(shù)據(jù)才能被寫入 Buffer 。

Channel

在 NIO 中，Channel 和 Buffer 是相輔相成的，我們只能從 Channel 讀取數(shù)據(jù)到 Buffer 中，或者從 Buffer 寫入數(shù)據(jù)到 Channle，如下圖：

Channel 類似于 OIO 中的流(Stream)，但是又有所區(qū)別：

• 流是單向的，但 Channel 是雙向的，可讀可寫。

• 流是阻塞的，但 Channle 可以異步讀寫。

• 流中的數(shù)據(jù)可以選擇性的先讀到緩存中，而 Channel 的數(shù)據(jù)總是要先讀到一個 Buffer 中，或從 Buffer 中寫入，如上圖。

NIO 中通過 Channel 封裝了對數(shù)據(jù)源的操作，通過 Channel 我們可以操作數(shù)據(jù)源，但是又不必關(guān)注數(shù)據(jù)源的具體物理結(jié)構(gòu)，這個數(shù)據(jù)源可以是文件，也可以是socket。

Channel 的接口定義如下：

public interface Channel extends Closeable {

    public boolean isOpen();

    public void close() throws IOException;
}

Channel 接口僅定義兩個方法：

• isOpen()：Channel 是否打開

• close()：關(guān)閉 Channel

它的主要實現(xiàn)有：

• FileChannel：文件通道，用于文件的數(shù)據(jù)讀寫。

• SocketChannel：套接字通道，能通過 TCP 讀寫網(wǎng)絡(luò)中的數(shù)據(jù)。

• ServerSocketChannel：服務(wù)器套接字通道，監(jiān)聽新進來的 TCP 連接，像 web 服務(wù)器那樣，對每一個新進來的連接都會創(chuàng)建一個 SocketChannel。

• DatagramChannel：數(shù)據(jù)報通道，能通過 UDP 讀寫網(wǎng)絡(luò)中的數(shù)據(jù)。

基本類圖如下：

下面就 FileChannel 做詳細(xì)介紹。

FileChannel

FileChannel 主要是用來讀寫和映射一個系統(tǒng)文件的 Channel，它是一個抽象類，具體由 FileChannelImpl 來實現(xiàn)。

定義如下：

package java.nio.channels;
 
public abstract class FileChannel
    extends AbstractInterruptibleChannel
    implements SeekableByteChannel, GatheringByteChannel, ScatteringByteChannel{
    /**
     * 初始化一個無參構(gòu)造器.
     */
    protected FileChannel() { }
 
    //打開或創(chuàng)建一個文件，返回一個文件通道來訪問文件
    public static FileChannel open(Path path,
                                   Set options,
                                   FileAttribute... attrs)
        throws IOException
    {
        FileSystemProvider provider = path.getFileSystem().provider();
        return provider.newFileChannel(path, options, attrs);
    }
 
    private static final FileAttribute[] NO_ATTRIBUTES = new FileAttribute[0];
 
    //打開或創(chuàng)建一個文件，返回一個文件通道來訪問文件
    public static FileChannel open(Path path, OpenOption... options)
        throws IOException
    {
        Set set = new HashSet(options.length);
        Collections.addAll(set, options);
        return open(path, set, NO_ATTRIBUTES);
    }
 
  //從這個通道讀入一個字節(jié)序列到給定的緩沖區(qū)
    public abstract int read(ByteBuffer dst) throws IOException;
 
    //從這個通道讀入指定開始位置和長度的字節(jié)序列到給定的緩沖區(qū)
    public abstract long read(ByteBuffer[] dsts, int offset, int length)
        throws IOException;
 
    /**
     * 從這個通道讀入一個字節(jié)序列到給定的緩沖區(qū)
     */
    public final long read(ByteBuffer[] dsts) throws IOException {
        return read(dsts, 0, dsts.length);
    }
 
    /**
     * 從給定的緩沖區(qū)寫入字節(jié)序列到這個通道
     */
    public abstract int write(ByteBuffer src) throws IOException;
 
    /**
     * 從給定緩沖區(qū)的子序列向該信道寫入字節(jié)序列
     */
    public abstract long write(ByteBuffer[] srcs, int offset, int length)
        throws IOException;
 
    /**
     * 從給定的緩沖區(qū)寫入字節(jié)序列到這個通道
     */
    public final long write(ByteBuffer[] srcs) throws IOException {
        return write(srcs, 0, srcs.length);
    }
 
    /**
     * 返回通道讀寫緩沖區(qū)中的開始位置
     */
    public abstract long position() throws IOException;
 
    /**
     * 設(shè)置通道讀寫緩沖區(qū)中的開始位置
     */
    public abstract FileChannel position(long newPosition) throws IOException;
 
    /**
     * 返回此通道文件的當(dāng)前大小
     */
    public abstract long size() throws IOException;
 
    /**
     * 通過指定的參數(shù)size來截取通道的大小
     */
    public abstract FileChannel truncate(long size) throws IOException;
 
    /**
     * 強制將通道中的更新文件寫入到存儲設(shè)備（磁盤等）中
     */
    public abstract void force(boolean metaData) throws IOException;
 
    /**
     * 將當(dāng)前通道中的文件寫入到可寫字節(jié)通道中
   * position就是開始寫的位置，long就是寫的長度
     */
    public abstract long transferTo(long position, long count,
                                    WritableByteChannel target)
        throws IOException;
 
    /**
     * 將當(dāng)前通道中的文件寫入可讀字節(jié)通道中
   * position就是開始寫的位置，long就是寫的長度
     */
    public abstract long transferFrom(ReadableByteChannel src,
                                      long position, long count)
        throws IOException;
 
    /**
     * 從通道中讀取一系列字節(jié)到給定的緩沖區(qū)中
   * 從指定的讀取開始位置position處讀取
     */
    public abstract int read(ByteBuffer dst, long position) throws IOException;
 
    /**
     * 從給定的緩沖區(qū)寫入字節(jié)序列到這個通道
     * 從指定的讀取開始位置position處開始寫
     */
    public abstract int write(ByteBuffer src, long position) throws IOException;
 
 
    // -- Memory-mapped buffers --
 
    /**
     * 一個文件映射模式類型安全枚舉
     */
    public static class MapMode {
 
        //只讀映射模型
        public static final MapMode READ_ONLY
            = new MapMode("READ_ONLY");
 
        //讀寫映射模型
        public static final MapMode READ_WRITE
            = new MapMode("READ_WRITE");
 
        /**
         * 私有模式（復(fù)制在寫）映射
         */
        public static final MapMode PRIVATE
            = new MapMode("PRIVATE");
 
        private final String name;
 
        private MapMode(String name) {
            this.name = name;
        }
    }
 
    /**
     * 將該通道文件的一個區(qū)域直接映射到內(nèi)存中
     */
    public abstract MappedByteBuffer map(MapMode mode,
                                         long position, long size)
        throws IOException;
 
    /**
     * 獲取當(dāng)前通道文件的給定區(qū)域上的鎖
   * 區(qū)域就是從position處開始，size長度 
   * shared為true代表獲取共享鎖，false代表獲取獨占鎖
     */
    public abstract FileLock lock(long position, long size, boolean shared)
        throws IOException;
 
    /**
     * 獲取當(dāng)前通道文件上的獨占鎖
     */
    public final FileLock lock() throws IOException {
        return lock(0L, Long.MAX_VALUE, false);
    }
 
    /**
     * 嘗試獲取給定的通道文件區(qū)域上的鎖
     * 區(qū)域就是從position處開始，size長度 
   * shared為true代表獲取共享鎖，false代表獲取獨占鎖
     */
    public abstract FileLock tryLock(long position, long size, boolean shared)
        throws IOException;
 
    /**
     * 嘗試獲取當(dāng)前通道文件上的獨占鎖
     */
    public final FileLock tryLock() throws IOException {
        return tryLock(0L, Long.MAX_VALUE, false);
    }
 
}

打開 FileChannel

在使用 FileChannle 之前我們必須要先打開它，但是我們無法直接打開一個 FileChannel，需要通過使用一個 InputStream、OutputStream、RandomAcessFile 來獲取一個 FileChannel 實例，如下:

RandomAccessFile accessFile = new RandomAccessFile("/Users/chenssy/Documents/FileChannel.txt","rw");
FileChannel fileChannel = accessFile.getChannel();

調(diào)用 getChannel() 即可獲取 FileChannel 實例，源碼如下：

public final FileChannel getChannel() {
    synchronized (this) {
        if (channel == null) {
            channel = FileChannelImpl.open(fd, path, true, rw, this);
        }
        return channel;
    }
}

getChnnel() 方法很簡單，直接調(diào)用 FileChannelImpl 的靜態(tài)方法 open()：

public static FileChannel open(Path path,
        Set options,
        FileAttribute... attrs) throws IOException{
    FileSystemProvider provider = path.getFileSystem().provider();
    return provider.newFileChannel(path, options, attrs);
}

從 FileChannel 讀數(shù)據(jù)

調(diào)用 FileChannel 的 read() 方法即可從 FileChannel 中獲取數(shù)據(jù)，當(dāng)然不是直接獲取，而是需要先寫入到 Buffer 中，所以調(diào)用 read() 之前，我們需要分配一個 Buffer，然后調(diào)用 read() ，該方法返回 int 表示有多少數(shù)據(jù)讀取到了 Buffer 中了，如果返回 -1 表示已經(jīng)到文件末尾了。

ByteBuffer buffer = ByteBuffer.allocate(1024);
int readCount = fileChannel.read(buffer);

FileChannel 僅定義了方法，具體實現(xiàn)在 FileChannelImpl，如下：

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 {
                // 通過IOUtil.read實現(xiàn)
                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);
        }
    }
}

• 首先確保該 Channel 是打開的

• 然后加鎖，主要是因為寫入緩沖區(qū)需要保證線程安全

• 最后通過 IOUtils.read() 實現(xiàn)

static int read(FileDescriptor fd, ByteBuffer dst, long position, NativeDispatcher nd) throws IOException
{
    // 1 申請一塊臨時堆外DirectByteBuffer
    ByteBuffer bb = Util.getTemporaryDirectBuffer(dst.remaining());
    try {
        // 2 先往DirectByteBuffer寫入數(shù)據(jù)，提高效率
        int n = readIntoNativeBuffer(fd, bb, position, nd);
        bb.flip();
        if (n > 0)
            // 3 再拷貝到傳入的buffer
            dst.put(bb);
        return n;
    } finally {
        Util.offerFirstTemporaryDirectBuffer(bb);
    }
}

• 首先申請一塊臨時的堆外 DirectByteBuffer

• 然后先往 DirectByteBuffer 寫入數(shù)據(jù)，因為這樣能夠提高效率，為什么會提高效率，我們后文分析。

• 最后拷貝到 ByteBuffer 中

寫數(shù)據(jù)到 FileChannel

read()方法是從 FileChannel 中讀取數(shù)據(jù)，那 write()方法則是從 ByteBuffer中讀取數(shù)據(jù)寫入到 Channel 中。調(diào)用 write() 需要先申請一個 ByteBuffer ，如下:

ByteBuffer buffer = ByteBuffer.allocate(1024);
fileChannel.write(buffer);

同樣，實現(xiàn)是在 FileChannelImpl 中。

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() 方法實現(xiàn)一模一樣，先確定該 Channel 是打開的，然后加鎖，最后調(diào)用 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);

   int pos = src.position();
   int lim = src.limit();
   assert (pos <= lim);
   int rem = (pos <= lim ? lim - pos : 0);
   // 2 否則構(gòu)造一塊跟傳入緩沖區(qū)一樣大小的DirectBuffer
   ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);
   try {
       bb.put(src);
       bb.flip();
       src.position(pos);

       // 3 調(diào)用writeFromNativeBuffer讀取
       int n = writeFromNativeBuffer(fd, bb, position, nd);
       if (n > 0) {
           // now update src
           src.position(pos + n);
       }
       return n;
   } finally {
       Util.offerFirstTemporaryDirectBuffer(bb);
   }
}

• 首先判斷傳入的 Buffer 是否為 DirectBuffer，如果是的話，就直接寫入

• 否則則構(gòu)造一塊跟傳入 Buffer 一樣大小的 DirectBuffer

• 最后調(diào)用 writeFromNativeBuffer()

關(guān)閉 FileChannel

保持好習(xí)慣，用完了一定要記得關(guān)閉：close()。

public final void close() throws IOException {
    synchronized (closeLock) {
        if (!open)
            return;
        open = false;
        implCloseChannel();
    }
}

調(diào)用 implCloseChannel() 釋放 Channel。

protected void implCloseChannel() throws IOException {
    // 釋放文件鎖
    if (fileLockTable != null) {
        for (FileLock fl: fileLockTable.removeAll()) {
            synchronized (fl) {
                if (fl.isValid()) {
                    //釋放鎖
                    nd.release(fd, fl.position(), fl.size());
                    ((FileLockImpl)fl).invalidate();
                }
            }
        }
    }
    // 通知當(dāng)前通道所有被阻塞線程
    threads.signalAndWait();
    if (parent != null) {
        ((java.io.Closeable)parent).close();
    } else {
        nd.close(fd);
    }
}

關(guān)閉 FileChannel 時，需要釋放所有鎖和文件流。

示例

讀數(shù)據(jù)

public static void main(String[] args) throws Exception {
    RandomAccessFile accessFile = new RandomAccessFile("/Users/chenssy/Documents/FileChannel.txt","rw");
    FileChannel fileChannel = accessFile.getChannel();

    ByteBuffer buffer = ByteBuffer.allocate(1024);
    fileChannel.read(buffer);
    System.out.println(new String(buffer.array()));
    fileChannel.close();
}

運行結(jié)果：

寫數(shù)據(jù)

public static void main(String[] args) throws Exception {
    String fileContent = "這是 chenssy 的 死磕 Java 系列中的文章....";
    RandomAccessFile accessFile = new RandomAccessFile("/Users/chenssy/Documents/FileChannel.txt","rw");
    FileChannel fileChannel = accessFile.getChannel();

    ByteBuffer buffer = ByteBuffer.allocate(1024);
    buffer.put(fileContent.getBytes("UTF-8"));
    buffer.flip();
    fileChannel.write(buffer);
    fileChannel.close();
}

運行結(jié)果：

參考資料

• Java NIO系列教程（七） FileChannel