Segmented Log 分段日志

Posted 2021-07-07 BBinChina

Martin Flower Segmented Log

我的使用场景：
在自研分布式流批存储系统时，元数据居于commit log的方式导致日志过大时，采用了分段日志的处理方式，下文内容翻译至马丁的文章，同时结合了自己遇到的问题。

Segmented Log 分段日志的思路是将原本单一的commit log切分成多个固定大小的小文件，我采用的是64M大小块，太大的话影响重启后读取，太小又造成频繁IO。

在什么时候切分文件呢? 基于commit log 思路，我们需要在写日记之前进行文件大小判断进行切分，这个文件大小状态在内存中进行维护，伪码如下

//持久化命令时
public Long writeEntry(WALEntry entry) {
	//检测是否需要切分
    maybeRoll();
    //commit log
    return openSegment.writeEntry(entry);
}

private void maybeRoll() {
	//判断设置的文件大小
    if (openSegment.
            size() >= config.getMaxLogSize()) {
            //刷盘
        openSegment.flush();
        //记录存储的分段日志
        sortedSavedSegments.add(openSegment);
  		//获取新文件句柄可进行写入
        long lastId = openSegment.getLastLogEntryId();
        openSegment = WALSegment.open(lastId, config.getWalDir());
    }
}

当我们读取日志时，对于给定的日记逻辑位置，需要映射到文件的物理存储位置，可以采用以下思路做映射。

• 每个日志文件名由设定的文件前缀和日志偏移量(或日志序列号)生成。
• 每个日志序号分为两个部分，即文件名称和事务偏移量。

public static String createFileName(Long startIndex) {
    return logPrefix + "_" + startIndex + logSuffix;
}

public static Long getBaseOffsetFromFileName(String fileName) {
    String[] nameAndSuffix = fileName.split(logSuffix);
    String[] prefixAndOffset = nameAndSuffix[0].split("_");
    if (prefixAndOffset[0].equals(logPrefix))
        return Long.parseLong(prefixAndOffset[1]);

    return -1l;
}

那么在读取操作时，分为两个步骤。对于给定的偏移量(或事务id)，将标识日志段，并从后续日志段读取所有日志记录。

public List<WALEntry> readFrom(Long startIndex) {
	//读取索引位置开始的日志（分段多个日志）
    List<WALSegment> segments = getAllSegmentsContainingLogGreaterThan(startIndex);
    return readWalEntriesFrom(startIndex, segments);
}
private List<WALSegment> getAllSegmentsContainingLogGreaterThan(Long startIndex) {
    List<WALSegment> segments = new ArrayList<>();
    //因为是递增顺序写的，所以当获取到段的基础偏移量小于索引时，既表示已获取获得所有日志条目多于startIndex的段
    for (int i = sortedSavedSegments.size() - 1; i >= 0; i--) {
        WALSegment walSegment = sortedSavedSegments.get(i);
        segments.add(walSegment);

		//倒序下的日志文件其偏移量小于索引值，不需要在读取更小的段文件了
        if (walSegment.getBaseOffset() <= startIndex) {
            break;
        }
    }

	//内存里的段
    if (openSegment.getBaseOffset() <= startIndex) {
        segments.add(openSegment);
    }

    return segments;
}