【死磕 NIO】— 深入分析Channel和FileChannel
大家好，我是大明哥，這次我們來(lái)看看NIO的第二個(gè)組件：Channel。
上篇文章[【死磕 NIO】— 深入分析Buffer]介紹了 NIO 中的 Buffer，Buffer 我們可以認(rèn)為他是裝載數(shù)據(jù)的容器，有了容器，還需要傳輸數(shù)據(jù)的通道才能完成數(shù)據(jù)的傳輸，這個(gè)通道就是今天要介紹的 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ù)總是要先讀到一個(gè) Buffer 中，或從 Buffer 中寫入，如上圖。
NIO 中通過(guò) Channel 封裝了對(duì)數(shù)據(jù)源的操作，通過(guò) Channel 我們可以操作數(shù)據(jù)源，但是又不必關(guān)注數(shù)據(jù)源的具體物理結(jié)構(gòu)，這個(gè)數(shù)據(jù)源可以是文件，也可以是socket。
Channel 的接口定義如下：
public interface Channel extends Closeable {
public boolean isOpen();
public void close() throws IOException;
}
Channel 接口僅定義兩個(gè)方法：
isOpen()：Channel 是否打開close()：關(guān)閉 Channel
它的主要實(shí)現(xiàn)有：
FileChannel：文件通道，用于文件的數(shù)據(jù)讀寫。
SocketChannel：套接字通道，能通過(guò) TCP 讀寫網(wǎng)絡(luò)中的數(shù)據(jù)。
ServerSocketChannel：服務(wù)器套接字通道，監(jiān)聽(tīng)新進(jìn)來(lái)的 TCP 連接，像 web 服務(wù)器那樣，對(duì)每一個(gè)新進(jìn)來(lái)的連接都會(huì)創(chuàng)建一個(gè) SocketChannel。
DatagramChannel：數(shù)據(jù)報(bào)通道，能通過(guò) UDP 讀寫網(wǎng)絡(luò)中的數(shù)據(jù)。
基本類圖如下：
下面就 FileChannel 做詳細(xì)介紹。
FileChannel
FileChannel 主要是用來(lái)讀寫和映射一個(gè)系統(tǒng)文件的 Channel，它是一個(gè)抽象類，具體由 FileChannelImpl 來(lái)實(shí)現(xiàn)。
定義如下：
package java.nio.channels;
public abstract class FileChannel
extends AbstractInterruptibleChannel
implements SeekableByteChannel, GatheringByteChannel, ScatteringByteChannel{
/**
* 初始化一個(gè)無(wú)參構(gòu)造器.
*/
protected FileChannel() { }
//打開或創(chuàng)建一個(gè)文件，返回一個(gè)文件通道來(lái)訪問(wèn)文件
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)建一個(gè)文件，返回一個(gè)文件通道來(lái)訪問(wèn)文件
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);
}
//從這個(gè)通道讀入一個(gè)字節(jié)序列到給定的緩沖區(qū)
public abstract int read(ByteBuffer dst) throws IOException;
//從這個(gè)通道讀入指定開始位置和長(zhǎng)度的字節(jié)序列到給定的緩沖區(qū)
public abstract long read(ByteBuffer[] dsts, int offset, int length)
throws IOException;
/**
* 從這個(gè)通道讀入一個(gè)字節(jié)序列到給定的緩沖區(qū)
*/
public final long read(ByteBuffer[] dsts) throws IOException {
return read(dsts, 0, dsts.length);
}
/**
* 從給定的緩沖區(qū)寫入字節(jié)序列到這個(gè)通道
*/
public abstract int write(ByteBuffer src) throws IOException;
/**
* 從給定緩沖區(qū)的子序列向該信道寫入字節(jié)序列
*/
public abstract long write(ByteBuffer[] srcs, int offset, int length)
throws IOException;
/**
* 從給定的緩沖區(qū)寫入字節(jié)序列到這個(gè)通道
*/
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;
/**
* 通過(guò)指定的參數(shù)size來(lái)截取通道的大小
*/
public abstract FileChannel truncate(long size) throws IOException;
/**
* 強(qiáng)制將通道中的更新文件寫入到存儲(chǔ)設(shè)備（磁盤等）中
*/
public abstract void force(boolean metaData) throws IOException;
/**
* 將當(dāng)前通道中的文件寫入到可寫字節(jié)通道中
* position就是開始寫的位置，long就是寫的長(zhǎng)度
*/
public abstract long transferTo(long position, long count,
WritableByteChannel target)
throws IOException;
/**
* 將當(dāng)前通道中的文件寫入可讀字節(jié)通道中
* position就是開始寫的位置，long就是寫的長(zhǎng)度
*/
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é)序列到這個(gè)通道
* 從指定的讀取開始位置position處開始寫
*/
public abstract int write(ByteBuffer src, long position) throws IOException;
// -- Memory-mapped buffers --
/**
* 一個(gè)文件映射模式類型安全枚舉
*/
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;
}
}
/**
* 將該通道文件的一個(gè)區(qū)域直接映射到內(nèi)存中
*/
public abstract MappedByteBuffer map(MapMode mode,
long position, long size)
throws IOException;
/**
* 獲取當(dāng)前通道文件的給定區(qū)域上的鎖
* 區(qū)域就是從position處開始，size長(zhǎng)度
* shared為true代表獲取共享鎖，false代表獲取獨(dú)占鎖
*/
public abstract FileLock lock(long position, long size, boolean shared)
throws IOException;
/**
* 獲取當(dāng)前通道文件上的獨(dú)占鎖
*/
public final FileLock lock() throws IOException {
return lock(0L, Long.MAX_VALUE, false);
}
/**
* 嘗試獲取給定的通道文件區(qū)域上的鎖
* 區(qū)域就是從position處開始，size長(zhǎng)度
* shared為true代表獲取共享鎖，false代表獲取獨(dú)占鎖
*/
public abstract FileLock tryLock(long position, long size, boolean shared)
throws IOException;
/**
* 嘗試獲取當(dāng)前通道文件上的獨(dú)占鎖
*/
public final FileLock tryLock() throws IOException {
return tryLock(0L, Long.MAX_VALUE, false);
}
}
打開 FileChannel
在使用 FileChannle 之前我們必須要先打開它，但是我們無(wú)法直接打開一個(gè) FileChannel，需要通過(guò)使用一個(gè) InputStream、OutputStream、RandomAcessFile 來(lái)獲取一個(gè) FileChannel 實(shí)例，如下:
RandomAccessFile accessFile = new RandomAccessFile("/Users/chenssy/Documents/FileChannel.txt","rw");
FileChannel fileChannel = accessFile.getChannel();
調(diào)用 getChannel() 即可獲取 FileChannel 實(shí)例，源碼如下：
public final FileChannel getChannel() {
synchronized (this) {
if (channel == null) {
channel = FileChannelImpl.open(fd, path, true, rw, this);
}
return channel;
}
}
getChnnel() 方法很簡(jiǎn)單，直接調(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() 之前，我們需要分配一個(gè) Buffer，然后調(diào)用 read() ，該方法返回 int 表示有多少數(shù)據(jù)讀取到了 Buffer 中了，如果返回 -1 表示已經(jīng)到文件末尾了。
ByteBuffer buffer = ByteBuffer.allocate(1024);
int readCount = fileChannel.read(buffer);
FileChannel 僅定義了方法，具體實(shí)現(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 {
// 通過(guò)IOUtil.read實(shí)現(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 是打開的
然后加鎖，主要是因?yàn)閷懭刖彌_區(qū)需要保證線程安全
最后通過(guò)
IOUtils.read()實(shí)現(xiàn)
static int read(FileDescriptor fd, ByteBuffer dst, long position, NativeDispatcher nd) throws IOException
{
// 1 申請(qǐng)一塊臨時(shí)堆外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);
}
}
首先申請(qǐng)一塊臨時(shí)的堆外 DirectByteBuffer
然后先往 DirectByteBuffer 寫入數(shù)據(jù)，因?yàn)檫@樣能夠提高效率，為什么會(huì)提高效率，我們后文分析。
最后拷貝到 ByteBuffer 中
寫數(shù)據(jù)到 FileChannel
read()方法是從 FileChannel 中讀取數(shù)據(jù)，那 write()方法則是從 ByteBuffer中讀取數(shù)據(jù)寫入到 Channel 中。調(diào)用 write() 需要先申請(qǐng)一個(gè) ByteBuffer ，如下:
ByteBuffer buffer = ByteBuffer.allocate(1024);
fileChannel.write(buffer);
同樣，實(shí)現(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() 方法實(shí)現(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í)，需要釋放所有鎖和文件流。
示例
讀數(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();
}
運(yùn)行結(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();
}
運(yùn)行結(jié)果：
參考資料
Java NIO系列教程（七） FileChannel