FlowFile存储库原理

/**
 * 实现必须是线程安全的
 */
public interface FlowFileRepository extends Closeable {
    /**
     * 初始化内容存储库，向其提供要与内容声明进行交互的ContentClaimManager
     */
    void initialize(ResourceClaimManager claimManager) throws IOException;
    long getStorageCapacity() throws IOException;
    long getUsableStorageSpace() throws IOException;
    String getFileStoreName();

    /**
     * 使用给定的RepositoryRecords更新存储库。
     */
    void updateRepository(Collection<RepositoryRecord> records) throws IOException;

    /**
     * 加载在存储库中找到的所有流文件，建立内容声明及其引用计数
     *
     * @param queueProvider 将FlowFiles放入其中的FlowFile Queues的提供者
     *
     * @return 最高流文件标识符的索引
     */
    long loadFlowFiles(QueueProvider queueProvider) throws IOException;

    /**
     * 搜索存储库以查找当前已将数据排队的所有FlowFile队列的ID
     * @return FlowFile排队的所有FlowFileQueue标识符的集合
     */
    Set<String> findQueuesWithFlowFiles(FlowFileSwapManager flowFileSwapManager) throws IOException;

    boolean isVolatile();

    /**
     * @return 按顺序创建FlowFile的下一个ID。
     */
    long getNextFlowFileSequence();

    /**
     * @return 存储库中当前存在的所有流文件的最大ID。
     */
    long getMaxFlowFileIdentifier() throws IOException;

    /**
     * 通知FlowFile信息库已将给定标识符标识为外部（交换出）FlowFile遇到的最大值。
     */
    void updateMaxFlowFileIdentifier(long maxId);

    /**
     * 更新存储库以指示给定的FlowFileRecords已交换出内存
     */
    void swapFlowFilesOut(List<FlowFileRecord> swappedOut, FlowFileQueue flowFileQueue, String swapLocation) throws IOException;

    /**
     * 更新存储库以指示将给定的FlowFileRecords交换到内存中
     *
     * @param swapLocation the location (e.g., a filename) from which FlowFiles
     * were recovered
     * @param flowFileRecords the records that were swapped in
     * @param flowFileQueue the queue that the FlowFiles belong to
     */
    void swapFlowFilesIn(String swapLocation, List<FlowFileRecord> flowFileRecords, FlowFileQueue flowFileQueue) throws IOException;

    boolean isValidSwapLocationSuffix(String swapLocationSuffix);

    default Map<ResourceClaim, Set<ResourceClaimReference>> findResourceClaimReferences(Set<ResourceClaim> resourceClaims, FlowFileSwapManager swapManager)
        throws IOException {
        return null;
    }
    default Set<ResourceClaim> findOrphanedResourceClaims() {
        return Collections.emptySet();
    }
}

nifi.flowfile.repository.implementation=org.apache.nifi.controller.repository.WriteAheadFlowFileRepository
nifi.flowfile.repository.wal.implementation=org.apache.nifi.wali.SequentialAccessWriteAheadLog
nifi.flowfile.repository.directory=./flowfile_repository
nifi.flowfile.repository.checkpoint.interval=20 secs
nifi.flowfile.repository.always.sync=false

@Override
public void updateRepository(final Collection<RepositoryRecord> records) throws IOException {
    // alwaysSync 对应 `nifi.flowfile.repository.always.sync`，默认为false
    updateRepository(records, alwaysSync);
}

private void updateRepository(final Collection<RepositoryRecord> records, final boolean sync) throws IOException {
    // 校验 除了DELETE CONTENTMISSING CLEANUP_TRANSIENT_CLAIMS record是需要destination的(即路由到哪个队列)
    for (final RepositoryRecord record : records) {
        if (record.getType() != RepositoryRecordType.DELETE && record.getType() != RepositoryRecordType.CONTENTMISSING
                && record.getType() != RepositoryRecordType.CLEANUP_TRANSIENT_CLAIMS && record.getDestination() == null) {
            throw new IllegalArgumentException("Record " + record + " has no destination and Type is " + record.getType());
        }
    }

    // 根据分区记录的类型是否为“CLEANUP_TRANSIENT_CLAIMS”进行记录。我们这样做是因为我们不想将这些类型的记录发送到预写日志。
    // CLEANUP_TRANSIENT_CLAIMS 那些回滚了的record
    final Map<Boolean, List<RepositoryRecord>> partitionedRecords = records.stream()
            .collect(Collectors.partitioningBy(record -> record.getType() == RepositoryRecordType.CLEANUP_TRANSIENT_CLAIMS));
    // 剩下的这些确实发生了的变化 要记录到日志里
    List<RepositoryRecord> recordsForWal = partitionedRecords.get(Boolean.FALSE);
    if (recordsForWal == null) {
        recordsForWal = Collections.emptyList();
    }
    // 序列化的record
    final List<SerializedRepositoryRecord> serializedRecords = new ArrayList<>(recordsForWal.size());
    recordsForWal.forEach(record -> serializedRecords.add(new LiveSerializedRepositoryRecord(record)));

    // 把序列化的record更新存储库。
    final int partitionIndex = wal.update(serializedRecords, sync);

    updateContentClaims(records, partitionIndex);
}

/**
 * 使用指定的记录更新存储库。集合不得包含具有相同ID的多个记录
 *
 * @param records the records to update
 * @param forceSync 指定存储库是否强制将缓冲区里的数据刷新到磁盘。如果为false，则数据可以存储在操作系统缓冲区中，这可以提高性能，但是如果断电或操作系统崩溃，则可能导致数据丢失
 * @return the index of the Partition that performed the update
 */
int update(Collection<T> records, boolean forceSync) throws IOException;

@Override
public int update(final Collection<T> records, final boolean forceSync) throws IOException {
    // 是否已经从持久态将FlowFile恢复到内存里(比如重启)
    if (!recovered) {
        throw new IllegalStateException("Cannot update repository until record recovery has been performed");
    }
    // 读锁 ReentrantReadWriteLock
    journalReadLock.lock();
    try {
        // WriteAheadJournal 使用给定的记录集更新journal日志
        journal.update(records, recordLookup);
        // 强制同步
        if (forceSync) {
            journal.fsync();
            syncListener.onSync(PARTITION_INDEX);
        }
        // WriteAheadSnapshot 更新快照
        snapshot.update(records);
    } finally {
        // 释放读锁
        journalReadLock.unlock();
    }
    return PARTITION_INDEX;
}

@Override
public void update(final Collection<T> records, final RecordLookup<T> recordLookup) throws IOException {
    // 记录日志之前写一些信息 比如LengthDelimitedJournal类名 version 序列化类名 version等等
    if (!headerWritten) {
        throw new IllegalStateException("Cannot update journal file " + journalFile + " because no header has been written yet.");
    }

    if (records.isEmpty()) {
        return;
    }
    // 检测一些状态 有问题直接抛出异常(比如磁盘满了 too many open files 等等)
    checkState();

    File overflowFile = null;
    // 从池子里获取一个ByteArrayDataOutputStream(streamPool自己实现的 内部是BlockingQueue)
    // ByteArrayDataOutputStream是NIFI自己写的，包装了ByteArrayOutputStream和DataOutputStream
    // 而ByteArrayOutputStream我们应该都知道了 就是干字节缓冲的活的 然后一批刷到其他的输出流里(写日志的FileOutputStream)
    // DataOutputStream用来装饰ByteArrayOutputStream 按照我们定义的格式写数据
    final ByteArrayDataOutputStream bados = streamPool.borrowObject();

    try {
        FileOutputStream overflowFileOut = null;

        try {
            DataOutputStream dataOut = bados.getDataOutputStream();
            for (final T record : records) {
                // 获取给定record的唯一ID
                final Object recordId = serde.getRecordIdentifier(record);
                // 返回具有给定标识符的Record->修改前的record
                final T previousRecordState = recordLookup.lookup(recordId);
                // 通过给定的DataOutputStream将编辑记录(新record 旧record 有点那个redo undo日志的意思)序列化到日志。
                serde.serializeEdit(previousRecordState, record, dataOut);

                final int size = bados.getByteArrayOutputStream().size();
                // 字节缓冲区里保存的日志超过了5M了，且支持溢出为overflow文件
                // 指示对writeExternalFileReference(File, DataOutputStream)的调用对于此实现是否有效; 
                // maxInHeapSerializationBytes 默认5*1024*1024 5M
                if (serde.isWriteExternalFileReferenceSupported() && size > maxInHeapSerializationBytes) {
                    // 将现有缓冲区里的日志数据写到overflow文件并充值缓冲区
                    ...
                }
            }
            // 将输出刷到字节缓冲区里(到这里为止，FlowFile变更记录刷到了字节缓冲区里了)
            dataOut.flush();
            ...
        } ...
        final ByteArrayOutputStream baos = bados.getByteArrayOutputStream();
        // BufferedOutputStream(指向了journal文件的FileOutputStream)
        final OutputStream out = getOutputStream();

        final long transactionId;
        // 至此以上是并发存FlowFile变更到缓冲区(多个ByteArrayOutputStream)
        // 同步锁，将缓冲区内的日志数据刷到BufferedOutputStream缓冲区里，BufferedOutputStream再刷到journal文件的FileOutputStream
        synchronized (this) {
            checkState();
            try {
                transactionId = currentTransactionId++;
                transactionCount++;

                transactionPreamble.clear();
                transactionPreamble.putLong(transactionId);
                transactionPreamble.putInt(baos.size());

                out.write(TRANSACTION_FOLLOWS);
                out.write(transactionPreamble.array());
                baos.writeTo(out);
                out.flush();
            } ...
}