第6-1 章 EditLogOutputStream

分享未结  0  570

李延 LV6 2022-03-08

悬赏：20积分

# 1、作用

在FsEditLog中对单个文件的抽象接口，当是单机时，操作的时本地文件，实现是：EditLogOutputStream。当是HA集群是，操作的是共享文件，实现是：QuorumOutputStream

# 2、EditLogOutputStream

抽象接口，与IO流相似，定义了一些文件写和刷新的操作。

```
abstract public void write(FSEditLogOp op) throws IOException;
```

```
abstract protected void flushAndSync(boolean durable) throws IOException;

/**
 * Flush data to persistent store.
 * Collect sync metrics.
 */
public void flush() throws IOException {
  flush(true);
}

public void flush(boolean durable) throws IOException {
  numSync++;
  long start = monotonicNow();
  flushAndSync(durable);
  long end = monotonicNow();
  totalTimeSync += (end - start);
}
```

其中 FSEditLogOp 封装了具体的日志操作，所以的日志操作都回被封装为FSEditLogOp对象。并有对应的序列化方法，write通过这个对象可以直接将日志序列化

# 3、EditLogFileOutputStream

本地文件的实现类，就是一个对于本地文件write操作的封装，但其中有一点：需要注意，它构建了一个缓冲区。对于缓冲区使用两个buffer。一个用来写数据，一个用来读取数据。

当我们写入数据时，会直接将数据写入到写的buffer 中，当需要刷新缓存时，将写buffer 负值到读buffer上，并将写buffer清空，这时候我们只需要刷新读buffer，而有新数据写入是，继续在写buffer 写数据即可。

EditLogFileOutputStream成员变量如下：

```
private File file; //具体操作的文件
private FileOutputStream fp; //操作文件的io流
private EditsDoubleBuffer doubleBuf; //缓冲区，两个缓冲原素，实现同时读写
```

## 3.1 构造函数：

```java
public EditLogFileOutputStream(Configuration conf, File name, int size)
    throws IOException {
  super();
  shouldSyncWritesAndSkipFsync = conf.getBoolean(
          DFSConfigKeys.DFS_NAMENODE_EDITS_NOEDITLOGCHANNELFLUSH,
          DFSConfigKeys.DFS_NAMENODE_EDITS_NOEDITLOGCHANNELFLUSH_DEFAULT);

file = name;
  doubleBuf = new EditsDoubleBuffer(size);
  RandomAccessFile rp;
  if (shouldSyncWritesAndSkipFsync) {
    rp = new RandomAccessFile(name, "rws");
  } else {
    rp = new RandomAccessFile(name, "rw");
  }
  try {
    fp = new FileOutputStream(rp.getFD()); // open for append
  } catch (IOException e) {
    IOUtils.closeStream(rp);
    throw e;
  }
  fc = rp.getChannel();
  fc.position(fc.size());
}
```

我们看到就是将传入的 文件进行保存，并创建了双缓存的EditsDoubleBuffer对象。

## 3.2 write

```java
@Override
public void write(FSEditLogOp op) throws IOException {
  doubleBuf.writeOp(op, getCurrentLogVersion());
}
```

我们看到写操作就是写入到缓存中。跟进

```java
  private TxnBuffer bufCurrent; // current buffer for writing
  private TxnBuffer bufReady; // buffer ready for flushing

public void writeOp(FSEditLogOp op, int logVersion) throws IOException {
  bufCurrent.writeOp(op, logVersion);
}
```

在doubleBuf 中，有两个写的时候是在bugCurrent中写。

## 3.3 flush

```
@Override
public void flushAndSync(boolean durable) throws IOException {
  if (fp == null) {
    throw new IOException("Trying to use aborted output stream");
  }
  if (doubleBuf.isFlushed()) {
    LOG.info("Nothing to flush");
    return;
  }
  preallocate(); // preallocate file if necessary
  doubleBuf.flushTo(fp);
  if (durable && !shouldSkipFsyncForTests && !shouldSyncWritesAndSkipFsync) {
    fc.force(false); // metadata updates not needed
  }
}
```

刷新也同样，就是将缓存中的数据刷新到文件中

回帖

消灭零回复