原创 吴就业 109 0 2021-02-26
本文为博主原创文章,未经博主允许不得转载。
本文链接:https://wujiuye.com/article/c05edbbffcfe47fb9715be5a68d03a9b
作者:吴就业
链接:https://wujiuye.com/article/c05edbbffcfe47fb9715be5a68d03a9b
来源:吴就业的网络日记
本文为博主原创文章,未经博主允许不得转载。
本篇文章写于2021年02月26日,从公众号|掘金|CSDN手工同步过来(博客搬家),本篇为原创文章。
本篇内容包括:
NIO
文件读写-FileChannel
MappedByteBuffer
读写文件在我印象中,似乎很少有关于文件操作的面试题,而大多数面试题都围绕着高并发、网络编程、RPC、数据库,但其实掌握文件操作也同等重要。只是我们很少会碰到需要操作文件的需求,毕竟百分之九十的工作都是依靠操作数据库、网络通信完成,而存储都被各类关系型数据库、分布式数据库、缓存、搜索引擎、甚至云存储替代了。
虽然偶尔我们也需要实现文件上传的接口,但文件上传一般都会选择存储到云端,顶多就临时转存一下,相信很多人会选择直接百度拷贝一份代码完成文件存储了事,甚至于都不关心是如何实现的。而多数的表格导出操作也都依赖一些现成的框架,以致于面向文件编程的重要性被弱化了。
如果我们去研究一些框架的底层源码,我们就能发现掌握文件操作其实也很重要。以RocketMQ
为例,RocketMQ
的消息存储并没有借用数据库,也没有借用其它第三方框架,仅仅是用文件存储。我很好奇,为什么没有面试题问RocketMQ
的消息存储实现原理。
我自己也开发过一些组件/框架/中间件,但由于文件操作这块知识太欠缺,首先想到的都是依赖一些第三方存储中间件/库实现,如Redis
、Mysql
、LevelDB
,这直接提升了框架的使用成本。所以我也一直知道掌握文件操作的重要性。
有时候我也在想,为什么部署Kafka
(旧版本)要部署一个Zookeeper
,而部署Zookeeper
的作用只是用于管理节点、消费者、实现Leader
选举。部署Zookeeper
为了保证Zookeeper
的可用性又要部署几个节点,这无疑增加了Kafka
的使用成本。所以当我看到Alibaba Sentinel
实现集群限流功能提供嵌入式模式时就很理解为什么要同时提供嵌入式部署和独立部署两种模式。
我去年开始着手自研一个分布式延迟调度中间件,其实核心功能早就实现了,也以嵌入式部署的方式在项目中支撑业务功能。但为了去掉依赖Redis
实现存储功能、第三方框架实现RPC
功能、广播机制实现Leader
选举功能,我才决定重新写一个。因此我用Raft
共识算法+LevelDB
(Key-Value存储库
)替代Redis
实现存储、基于Netty
自己封装RPC
框架、基于Raft
算法替代广播实现Leader
选举。这直接就降低了这款自研中间件的使用成本。而在实现Raft
算法的日记Appender
时,我又遇到了同样的槛,但这次我选择跨过去。
阿里开源的众多项目中,除RocketMQ
的消息存储使用文件存储外,Sentinel
存储资源指标数据统计也是使用文件存储,这两个框架在实现存储上都使用了同一种设计思想,即数据文件+索引文件。我在自研分布式延迟调度中间件中就借鉴了RocketMQ
与Sentinel
中的文件存储索引设计,数据文件存储日记,而索引文件存储日记ID
与日记在数据文件中的物理偏移量。
Sentinel
按精确到秒的时间戳存储索引,和时间戳是有序增长的,而且时间戳是long
类型占8个
字节,根据单个指标文件的最大大小,物理偏移量也正好可以是long
类型,因此每个索引占16
个字节。资源指标数据可能由于某段时间没有请求或者应用重启导致某些时间戳没有记录,但至少时间戳是单调递增的,因此我们只需要采用简单的折半查找就能快速定位到索引。由于Sentinel
资源指标数据收集不需要考虑高并发,这样的设计足以满足需求。
RocketMQ
需要提供可以通过key
或时间区间来查询消息的功能,因此RocketMQ
的索引存储实现相对Sentinel
较难。单个索引文件固定的文件大小约为400M
,一个索引文件可以保存2000W
个索引,索引文件的底层存储设计相当于是在文件系统中实现HashMap
结构,每个文件头存储了此文件存储的消息的最小时间戳和最大时间戳,这用于实现按时间区间搜索消息记录。消息key
的hash
值则作为索引项存放在索引文件中的物理偏移量,当然,还要加上文件头的大小,以及乘以单项索引占用的字节数。
你或许觉得,RocketMQ
与Sentinel
实现索引难在算法,的确,算法是灵魂。但软件的强大依然需要依赖硬件的支持,你是否考虑到,如何跳转到文件中的某个位置读取指定字节的数据,又如何改写文件中指定位置的数据?如何考虑并发读写问题,如何调优性能?文件的NIO
如何理解、如何使用MappedByteBuffer
提升性能以及原理是什么?
解读这几个问题是我学习文件读写的目的,也体现了掌握面向文件编程的重要性。虽然为了提升工作效率以及降低犯错率我们并不需要重复造轮子,但重复造轮子无疑是提升自身能力最高效的学习方法。
由于文件与目录的创建和删除较为简单,因此忽略这部分内容的介绍,我们重点学习文件的读写。
由于流是单向的,简单文件写可使用FileOutputStream
,而读文件则使用FileInputStream
。
任何数据输出到文件都是以字节为单位输出,包括图片、音频、视频。以图片为例,如果没有图片格式解析器,那么图片文件其实存储的就只是按某种格式存储的字节数据罢了。
FileOutputStream
指文件字节输出流,用于将字节数据输出到文件,仅支持顺序写入、支持以追加方式写入,但不支持在指定位置写入。
打开一个文件输出流并写入数据的示例代码如下。
public class FileOutputStreamStu{
public void testWrite(byte[] data) throws IOException {
try(FileOutputStream fos = new FileOutputStream("/tmp/test.file",true)) {
fos.write(data);
fos.flush();
}
}
}
注意,如果不指定追加方式打开流,new FileOutputStream
时会导致文件内容被清空,而FileOutputStream
的默认构建函数是以非追加模式打开流的。
FileOutputStream
的参数1
为文件名,参数2
为是否以追加模式打开流,如果为true
,则字节将写入文件的尾部而不是开头。
调用flush
方法目的是在流关闭之前清空缓冲区数据,实际上使用FileOutputStream
并不需要调用flush
方法,此处的刷盘指的是将缓存在JVM
内存中的数据调用系统函数write
写入。如BufferedOutputStream
,在调用BufferedOutputStream
方法时,如果缓存未满,实际上是不会调用系统函数write
的,如下代码所示。
public class BufferedOutputStream extends FilterOutputStream {
public synchronized void write(byte b[], int off, int len) throws IOException {
if (len >= buf.length) {
flushBuffer();
out.write(b, off, len);
return;
}
if (len > buf.length - count) {
flushBuffer();
}
System.arraycopy(b, off, buf, count, len); // 只写入缓存
count += len;
}
}
FileInputStream
指文件字节输入流,用于将文件中的字节数据读取到内存中,仅支持顺序读取,不可跳跃读取。
打开一个文件输入流读取数据的案例代码如下。
public class FileInputStreamStu{
public void testRead() throws IOException {
try (FileInputStream fis = new FileInputStream("/tmp/test/test.log")) {
byte[] buf = new byte[1024];
int realReadLength = fis.read(buf);
}
}
}
其中buf
数组中下标从0
到realReadLength
的字节数据就是实际读取的数据,如果realReadLength
返回-1
,则说明已经读取到文件尾并且未读取到任何数据。
当然,我们还可以一个字节一个字节的读取,如下代码所示。
public class FileInputStreamStu{
public void testRead() throws IOException {
try (FileInputStream fis = new FileInputStream("/tmp/test/test.log")) {
int byteData = fis.read(); // 返回值取值范围:[-1,255]
if (byteData == -1) {
return; // 读取到文件尾了
}
byte data = (byte) byteData;
// data为读取到的字节数据
}
}
}
至于读取到的字节数据如何使用就需要看你文件中存储的是什么数据了。
如果整个文件存储的是一张图片,那么需要将整个文件读取完,再按格式解析成图片,而如果整个文件是配置文件,则可以一行一行读取,遇到\n
换行符则为一行,代码如下。
public class FileInputStreamStu{
@Test
public void testRead() throws IOException {
try (FileInputStream fis = new FileInputStream("/tmp/test/test.log")) {
ByteBuffer buffer = ByteBuffer.allocate(1024);
int byteData;
while ((byteData = fis.read()) != -1) {
if (byteData == '\n') {
buffer.flip();
String line = new String(buffer.array(), buffer.position(), buffer.limit());
System.out.println(line);
buffer.clear();
continue;
}
buffer.put((byte) byteData);
}
}
}
}
Java
基于InputStream
、OutputStream
还提供了很多的API
方便读写文件,如BufferedReader
,但如果懒得去记这些API
的话,只需要记住FileInputStream
与FileOutputStream
就够了。
RandomAccessFile
相当于是FileInputStream
与FileOutputStream
的封装结合,即可以读也可以写,并且RandomAccessFile
支持移动到文件指定位置处开始读或写。
RandomAccessFile
的使用如下。
public class RandomAccessFileStu{
public void testRandomWrite(long index,long offset){
try (RandomAccessFile randomAccessFile = new RandomAccessFile("/tmp/test.idx", "rw")) {
randomAccessFile.seek(index * indexLength());
randomAccessFile.write(toByte(index));
randomAccessFile.write(toByte(offset));
}
}
}
RandomAccessFile
构建方法:参数1
为文件路径,参数2
为模式,'r'
为读,'w'
为写;seek
方法:在linux
、unix
操作系统下就是调用系统的lseek
函数。RandomAccessFile
的seek
方法通过调用native
方法实现,源码如下。
JNIEXPORT void JNICALL
Java_java_io_RandomAccessFile_seek0(JNIEnv *env,
jobject this, jlong pos) {
FD fd;
fd = GET_FD(this, raf_fd);
if (fd == -1) {
JNU_ThrowIOException(env, "Stream Closed");
return;
}
if (pos < jlong_zero) {
JNU_ThrowIOException(env, "Negative seek offset");
}
// #define IO_Lseek lseek
else if (IO_Lseek(fd, pos, SEEK_SET) == -1) {
JNU_ThrowIOExceptionWithLastError(env, "Seek failed");
}
}
Java_java_io_RandomAccessFile_seek0
函数的参数1
表示RandomAccessFile
对象,参数2
表示偏移量。函数中调用的IO_Lseek
方法实际是操作系统的lseek
方法。
RandomAccessFile
提供的读、写、指定偏移量其实都是通过调用操作系统函数完成的,包括前面介绍的文件输入流和文件输出流也不例外。
Channel
(通道)表示IO
源与目标打开的连接,Channel
类似于传统的流,但Channel
本身不能直接访问数据,只能与Buffer
进行交互。Channel
(通道)主要用于传输数据,从缓冲区的一侧传到另一侧的实体(如File
、Socket
),支持双向传递。
正如SocketChannel
是客户端与服务端通信的通道,FileChannel
就是我们读写文件的通道。FileChannel
是线程安全的,也就是一个FileChannel
可以被多个线程使用。对于多线程操作,同时只会有一个线程能对该通道所在文件进行修改。如果需要确保多线程的写入顺序,就必须要转为队列写入。
FileChannel
可通过FileOutputStream
、FileInputStream
、RandomAccessFile
获取,也可以通过FileChannel#open
方法打开一个通道。
以通过FileOutputStream
获取FileChannel
为例,通过FileOutputStream
或RandomAccessFile
获取FileChannel
方法相同,代码如下。
public class FileChannelStu{
public void testGetFileCahnnel(){
try(FileOutputStream fos = new FileOutputStream("/tmp/test.log");
FileChannel fileChannel = fos.getChannel()){
// do....
}catch (IOException exception){
}
}
}
需要注意,通过FileOutputStream
获取的FileChannel
只能执行写操作,通过FileInputStream
获取的FileChannel
只能执行读操作,原因可查看getChannel
方法源码。
通过FileOutputStream
或FileInputStream
或RandomAccessFile
打开的FileChannel
,在流关闭时也会被关闭,可查看这几个类的close
方法源码。
若想要获取一个同时支持读和写的FileChannel
需要通过open
方法打开,代码如下。
public class FileChannelStu{
public void testOpenFileCahnnel(){
FileChannel channel = FileChannel.open(
Paths.get(URI.create("file:" + rootPath + "/" + postion.fileName)),
StandardOpenOption.READ,StandardOpenOption.WRITE);
// do....
channel.close();
}
}
open
方法第二个变长参数传StandardOpenOption.READ
和StandardOpenOption.WRITE
即可打开一个双向读写的通道。
FileChannel
允许对文件加锁,文件锁是进程级别的,不是线程级别的,文件锁可以解决多个进程并发访问、修改同一个文件的问题。文件锁会被当前进程持有,一旦获取到文件锁就要调用一次release
释放锁,当关闭对应的FileChannel
对象时或当前JVM
进程退出时,锁也会自动被释锁。
文件锁的使用案例代码如下。
public class FileChannelStu{
public void testFileLock(){
FileChannel channel = this.channel;
FileLock fileLock = null;
try {
fileLock = channel.lock();// 获取文件锁
// 执行写操作
channel.write(...);
channel.write(...);
} finally {
if (fileLock != null) {
fileLock.release(); // 释放文件锁
}
}
}
}
当然,只要我们能确保同时只有一个进程对文件执行写操作,那么就不需要锁文件。RocketMQ
也并没有使用文件锁,因为每个Broker
有自己数据目录,即使一台机器上部署多个Broker
也不会有多个进程对同一个日记文件操作的情况。
上面例子去掉文件锁后代码如下。
public class FileChannelStu{
public void testWrite(){
FileChannel channel = this.channel;
channel.write(...);
channel.write(...);
}
}
这里还存在一个问题,就是并发写数据问题。虽然FileChannel
是线程安全的,但两次write
并不是原子性操作,如果要确保两次write
是连续写入的,还必须要加锁。在RocketMQ
中,通过引用计数器替代了锁。
FileChannel
提供的force
方法用于刷盘,即调用操作系统的fsync
函数,使用如下。
public class FileChannelStu{
public void closeChannel(){
this.channel.force(true);
this.channel.close();
}
}
force
方法的参数表示除强制写入内容更改外,文件元数据的更改是否也强制写入。后面使用MappedByteBuffer
时,可直接使用MappedByteBuffer
的force
方法。
FileChannel
的force
方法最终调用的C
方法源码如下:
JNIEXPORT jint JNICALL
Java_sun_nio_ch_FileDispatcherImpl_force0(JNIEnv *env, jobject this,
jobject fdo, jboolean md)
{
jint fd = fdval(env, fdo);
int result = 0;
if (md == JNI_FALSE) {
result = fdatasync(fd);
} else {
result = fsync(fd);
}
return handle(env, result, "Force failed");
}
参数md
对应调用force
方法传递的metaData
参数。
使用FileChannel
支持seek
(position
)到指定位置读或写数据,代码如下。
public class FileChannelStu{
public void testSeekWrite(){
FileChannel channel = this.channel;
synchronized (channel) {
channel.position(100);
channel.write(ByteBuffer.wrap(toByte(index)));
channel.write(ByteBuffer.wrap(toByte(offset)));
}
}
}
上述例子的作用是将指针移动到物理偏移量100byte
位置处,顺序写入index
和offset
。读取同理,代码如下。
public class FileChannelStu{
public void testSeekRead(){
FileChannel channel = this.channel;
synchronized (channel) {
channel.position(100);
ByteBuffer buffer = ByteBuffer.allocate(16);
int realReadLength = channel.read(buffer);
if(realReadLength==16){
long index = buffer.getLong();
long offset = buffer.getLong();
}
}
}
}
其中read
方法返回的是实际读取的字节数,如果返回-1
则代表已经是文件尾部了,没有剩余内容可读取。
MappedByteBuffer
是Java
提供的基于操作系统虚拟内存映射(MMAP
)技术的文件读写API
,底层不再通过read
、write
、seek
等系统调用实现文件的读写。
我们需要通过FileChannel#map
方法将文件的一个区域映射到内存中,代码如下。
public class MappedByteBufferStu{
@Test
public void testMappedByteBuffer() throws IOException {
FileChannel fileChannel = FileChannel.open(Paths.get(URI.create("file:/tmp/test/test.log")),
StandardOpenOption.WRITE, StandardOpenOption.READ);
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 4096);
fileChannel.close();
mappedByteBuffer.position(1024);
mappedByteBuffer.putLong(10000L);
mappedByteBuffer.force();
}
}
上面代码的功能是通过FileChannel
将文件[0~4096)
区域映射到内存中,调用FileChannel
的map
方法返回MappedByteBuffer
,在映射之后关闭通道,随后在指定位置处写入一个8
字节的long
类型整数,最后调用force
方法将写入数据从内存写回磁盘(刷盘)。
映射一旦建立了,就不依赖于用于创建它的文件通道,因此在创建MappedByteBuffer
之后我们就可以关闭通道了,对映射的有效性没有影响。
实际上将文件映射到内存比通过read
、write
系统调用方法读取或写入几十KB
的数据要昂贵,从性能的角度来看,MappedByteBuffer
适合用于将大文件映射到内存中,如上百M
、上GB
的大文件。
FileChannel
的map
方法有三个参数:
* MapMode
:映射模式,可取值有READ_ONLY
(只读映射)、READ_WRITE
(读写映射)、PRIVATE
(私有映射),READ_ONLY
只支持读,READ_WRITE
支持读写,而PRIVATE
只支持在内存中修改,不会写回磁盘;
* position
和size
:映射区域,可以是整个文件,也可以是文件的某一部分,单位为字节。
需要注意的是,如果FileChannel
是只读模式,那么map
方法的映射模式就不能指定为READ_WRITE
。如果文件是刚刚创建的,只要映射成功,文件的大小就会变成(0+position+size
)。
通过MappedByteBuffer
读取数据示例如下:
public class MappedByteBufferStu{
@Test
public void testMappedByteBufferOnlyRead() throws IOException {
FileChannel fileChannel = FileChannel.open(Paths.get(URI.create("file:/tmp/test/test.log")),
StandardOpenOption.READ);
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_ONLY, 0, 4096);
fileChannel.close();
mappedByteBuffer.position(1024);
long value = mappedByteBuffer.getLong();
System.out.println(value);
}
}
mmap
绕过了read
、write
系统函数调用,绕过了一次数据从内核空间到用户空间的拷贝,即实现零拷贝,MappedByteBuffer
使用直接内存而非JVM
的堆内存。
mmap
只是在虚拟内存分配了地址空间,只有在第一次访问虚拟内存的时候才分配物理内存。在mmap
之后,并没有将文件内容加载到物理页上,而是在虚拟内存中分配地址空间,当进程在访问这段地址时,通过查找页表,发现虚拟内存对应的页没有在物理内存中缓存则产生缺页中断,由内核的缺页异常处理程序处理,将文件对应内容以页为单位(4096
)加载到物理内存中。
由于物理内存是有限的,mmap
在写入数据超过物理内存时,操作系统会进行页置换,根据淘汰算法,将需要淘汰的页置换成所需的新页,所以mmap
对应的内存是可以被淘汰的,被淘汰的内存页如果是脏页(有过写操作修改页内容),则操作系统会先将数据回写磁盘再淘汰该页。
数据写过程如下:
* 1
.将需要写入的数据写到对应的虚拟内存地址;
* 2
.若对应的虚拟内存地址未对应物理内存,则产生缺页中断,由内核加载页数据到物理内存;
* 3
.数据被写入到虚拟内存对应的物理内存;
* 4
.在发生页淘汰或刷盘时由操作系统将脏页回写到磁盘。
RocketMQ
正是利用MappedByteBuffer
实现索引文件的读写,实现一个基于文件系统的HashMap
。
RocketMQ
在创建新的CommitLog
文件并通过FileChannel
获取MappedByteBuffer
时会做一次预热操作,即每个虚拟内存页(Page Cache
)都写入四个字节的0x00
,并强制刷盘将数据写到文件中。这个动作的用处是通过读写操作把MMAP
映射全部加载到物理内存中。并且在预热之后还做了一个锁住内存的操作,这是为了避免磁盘交换,防止操作系统把预热过的页临时保存到swap
区,防止程序再次读取交换出去的数据页时产生缺页中断。
声明:公众号、CSDN、掘金的曾用名:“Java艺术”,因此您可能看到一些早期的文章的图片有“Java艺术”的水印。
继 《Java文件的简单读写、随机读写、NIO读写与使用MappedByteBuffer读写》,本篇通过调用Linux OS文件操作系统函数实现copy命令以加深我们对Java文件读写底层实现的理解。
订阅
订阅新文章发布通知吧,不错过精彩内容!
输入邮箱,提交后我们会给您发送一封邮件,您需点击邮件中的链接完成订阅设置。