流

速览

Stream 模块以持久且并发的方式渲染批量文档，向单主机持久存储提供恰好一次的本地提交（跨主机的恰好一次是 Enterprise Stream 的边界）。它把工作拆分为两项清晰分离的职责：一个把已校验清单转换为字节（仅此而已）的渲染引擎，以及一组持久存储——committer、检查点、幂等、死信——它们安全地发布这些字节，并让一次运行在崩溃后无需重新发布已提交的输出即可恢复。

供应与许可

此能力随 NextPDF Pro（nextpdf/pro）提供，并通过 Pro 级许可信封激活。没有该授权的部署不会加载此能力的类。比较版本并获取授权。

没有单独的单项功能许可标志。并发度（worker 数量）、批大小、重试预算，以及存储后端（内存式与持久文件系统之别）都是运行时参数，而非许可开关。

安装

composer require nextpdf/pro:^3

代码位于 NextPDF\Pro\Stream 命名空间下。

概念概览

Stream 围绕一道冻结的接缝——NextPDF\Pro\Stream\Engine\RenderEngineInterface——组织起来，它把吞吐引擎与流语义分离开：

• 渲染引擎掌管并发与有界内存。 它通过 renderBatch() 渲染一个由预校验、预去重清单组成的窗口，并按输入顺序为每个清单返回一个 EngineRenderResult。关键在于，引擎就最终输出而言是无副作用的：它返回渲染后的字节加上它们的 sha-256 摘要，绝不写入最终的对象键。正是这种纯粹性使恰好一次投递成为可能。
• 流协作者掌管投递。 committer、检查点存储、幂等（去重）存储，以及死信存储决定字节落在何处、一次运行如何恢复、哪些工作属于重放，以及终止性失败将如何处理。

单个清单的渲染失败被报告为单条目的 Failed（或 Timeout）结果；它绝不会中止整个批次。批次信封始终成功，并附带逐条目的结果。

关键概念

渲染引擎与执行器

• InProcessRenderEngine 是同步、单进程的正确性基线。它在渲染之前，通过随附的 RenderManifestValidator 对每个清单进行失败关闭（fail-closed）校验，然后再交给 Core 的 SingleDocumentRenderer 渲染，因此一个坏清单会变成单条目失败（错误码 SPEC-MANIFEST-INVALID），而不会抵达渲染器。
• ConcurrentRenderEngine 把一个批次扇出给一个 RenderUnitExecutorInterface，并按单元索引恢复确定性的批次顺序。无论完成顺序如何，输出都与顺序渲染逐字节一致；缺失、重复或未知的完成项是硬失败，绝不会被悄悄丢弃。
• 执行器是并发接缝。InlineRenderUnitExecutor 是确定性基线；ProcessPoolRenderUnitExecutor 把一个批次分发到至多 N 个并行渲染的 php worker 子进程，然后收集并完整性校验它们的结果。

持久、无副作用的提交

OutputCommitterInterface::commit() 把渲染后的字节恰好一次地发布到其最终目的地：原子地（绝不会观察到任何部分对象）、幂等地（重新提交逐字节相同的内容不执行任何写入，并返回一个 idempotentReuse = true 的 CommitReceipt——是一张全新的回执，而非原始那张）、不静默覆盖（在没有 overwrite 的情况下向已占用的键写入有差异的字节会引发冲突），以及经完整性校验（committer 在写入前重算摘要）。LocalFilesystemCommitter 为本地文件系统实现了这一点。

检查点恢复

一个 RunCheckpoint 是一道持久屏障，记录一次运行已提交了多少条目，外加一份带键状态的快照。在恢复时，处理器快进越过已提交偏移量并恢复带键状态，因此运行途中的一次崩溃在恢复后无需重新发布已提交输出。FilesystemCheckpointStore 原子地持久化每一道屏障。

幂等去重、重试与死信

幂等存储是快速路径，它让处理器在渲染一个被重放的清单之前短路；committer 的摘要比对仍然是持久的恰好一次保证，因此丢失一条去重记录最坏只会导致一次被浪费的重新渲染，而该结果会被 committer 去重。RetryPolicy 为瞬时（超时）失败提供有界、确定性的指数退避；一个耗尽预算的作业会被捕获进一个 DeadLetterStoreInterface，而不会丢失。每个存储都随附一个内存变体（单次运行/测试范围）和一个持久文件系统变体。

持久性标记

状态可在进程重启后留存的存储实现 DurableCapability 标记。一次崩溃安全的运行要求每个协作者都是持久的，因此它会快速失败，而不会承诺一个内存式存储在重启后无法兑现的恰好一次语义。

代码示例 — 快速开始

渲染一个清单，并将其字节恰好一次地提交。引擎返回字节加摘要；committer 发布它们。

stream-quickstart.php

<?php

declare(strict_types=1);

use NextPDF\Manifest\OutputObjectKey;
use NextPDF\Manifest\Render\SingleDocumentRenderer;
use NextPDF\Manifest\RenderManifestBuilder;
use NextPDF\Manifest\TemplateRef;
use NextPDF\Pro\Stream\Commit\LocalFilesystemCommitter;
use NextPDF\Pro\Stream\Engine\InProcessRenderEngine;

$outputRoot = __DIR__ . '/out';
\is_dir($outputRoot) || \mkdir($outputRoot, 0o775, true);

// The engine renders bytes only — it never writes the final object.
$engine = new InProcessRenderEngine(SingleDocumentRenderer::standalone());

$target = OutputObjectKey::file('out', 'invoices/1001.pdf');

$manifest = RenderManifestBuilder::create('invoice-1001')
    ->withInlineInput('<h1>Invoice 1001</h1><p>Amount due: 42.00</p>')
    ->withTemplate(TemplateRef::html())
    ->withOutputKey($target)
    ->build();

$result = $engine->renderBatch([$manifest])[0];

// A durable committer publishes the rendered bytes exactly once.
$committer = new LocalFilesystemCommitter($outputRoot);

if ($result->isRendered()) {
    $receipt = $committer->commit($result->jobId, $target, $result->bytes, $result->sha256);
    echo $receipt->target->toUri(), ' (', $receipt->bytesWritten, " bytes)\n";
}

代码示例 — 生产

渲染一个批次，把超时路由到重试策略，并把终止性失败送入死信。提交拒绝覆盖有差异的字节，因此键冲突会被捕获并留存，而不会丢失。

stream-production.php

<?php

declare(strict_types=1);

use DateTimeImmutable;
use NextPDF\Manifest\OutputObjectKey;
use NextPDF\Manifest\Render\SingleDocumentRenderer;
use NextPDF\Manifest\RenderManifest;
use NextPDF\Manifest\RenderManifestBuilder;
use NextPDF\Manifest\TemplateRef;
use NextPDF\Pro\Stream\Commit\LocalFilesystemCommitter;
use NextPDF\Pro\Stream\Engine\EngineRenderStatus;
use NextPDF\Pro\Stream\Engine\InProcessRenderEngine;
use NextPDF\Pro\Stream\Exception\OutputCommitConflictException;
use NextPDF\Pro\Stream\Retry\DeadLetterRecord;
use NextPDF\Pro\Stream\Retry\InMemoryDeadLetterStore;
use NextPDF\Pro\Stream\Retry\RetryPolicy;

$outputRoot = __DIR__ . '/out';
\is_dir($outputRoot) || \mkdir($outputRoot, 0o775, true);

$engine     = new InProcessRenderEngine(SingleDocumentRenderer::standalone(), maxBatchSize: 64);
$committer  = new LocalFilesystemCommitter($outputRoot);
$deadLetter = new InMemoryDeadLetterStore();
$retry      = RetryPolicy::default(); // 3 attempts, 100ms base, 30s cap.

/**
 * Build one manifest and remember its output target for the commit stage.
 *
 * @return array{RenderManifest, OutputObjectKey}
 */
$makeJob = static function (string $jobId, string $html): array {
    $target   = OutputObjectKey::file('out', 'invoices/' . $jobId . '.pdf');
    $manifest = RenderManifestBuilder::create($jobId)
        ->withInlineInput($html)
        ->withTemplate(TemplateRef::html())
        ->withOutputKey($target)
        ->build();

    return [$manifest, $target];
};

/** @var array<non-empty-string, OutputObjectKey> $targets */
$targets   = [];
$manifests = [];
foreach (['inv-2001' => '<h1>2001</h1>', 'inv-2002' => '<h1>2002</h1>'] as $id => $html) {
    [$manifest, $target] = $makeJob($id, $html);
    $manifests[]  = $manifest;
    $targets[$id] = $target;
}

foreach ($engine->renderBatch($manifests) as $result) {
    // A timeout is transient — the policy decides whether to re-enqueue it.
    if ($result->status === EngineRenderStatus::Timeout && $retry->shouldRetry(1)) {
        // Re-enqueue on the caller's work queue after delayMsForAttempt(1) ms.
        continue;
    }

    if (!$result->isRendered()) {
        $deadLetter->add(new DeadLetterRecord(
            jobId: $result->jobId,
            idempotencyKeyValue: $result->jobId,
            attempts: $retry->maxAttempts,
            lastErrorCode: $result->errorCode ?? 'SPEC-RENDER-EXCEPTION',
            lastErrorMessage: $result->errorMessage ?? '',
            failedAt: new DateTimeImmutable(),
        ));

        continue;
    }

    try {
        // overwrite=false: identical bytes are an idempotent no-op; divergent
        // bytes to an occupied key raise SPEC-COMMIT-409 instead of clobbering.
        $receipt = $committer->commit(
            $result->jobId,
            $targets[$result->jobId],
            $result->bytes,
            $result->sha256,
        );
    } catch (OutputCommitConflictException $e) {
        $deadLetter->add(new DeadLetterRecord(
            jobId: $result->jobId,
            idempotencyKeyValue: $result->jobId,
            attempts: 1,
            lastErrorCode: $e->specCode(),
            lastErrorMessage: $e->getMessage(),
            failedAt: new DateTimeImmutable(),
        ));

        continue;
    }

    echo $receipt->idempotentReuse
        ? "reused {$receipt->target->toUri()}\n"
        : "committed {$receipt->target->toUri()}\n";
}

if ($deadLetter->count() > 0) {
    \fwrite(\STDERR, $deadLetter->count() . " job(s) dead-lettered\n");
}

何时使用

• 高吞吐量批量渲染，且吞吐量能从并发（进程池）执行中获益。
• 必须在崩溃后留存并恢复、且不重复发布输出的长时间运行。
• 必须保证每份渲染文档向其目标恰好一次投递的管线。

对于单份临时文档，请直接用 Writer 模块渲染；Stream 的价值在于持久、可恢复、并发的批次。

性能

吞吐量随 ProcessPoolRenderUnitExecutor 中的 worker 数量伸缩（受 maxWorkers 与 maxBatchSize 限定），同时引擎让渲染输出与顺序基线逐字节一致。一个墙钟超时为每个并行批次设置上限，因此一个挂起的 worker 不会永久阻塞。没有已发布的固定吞吐量数字；它取决于文档复杂度与主机并行度。请用有代表性的文档测量。

安全说明

清单在渲染之前经过失败关闭校验。committer 拒绝路径穿越、空字节、流包装器（stream-wrapper）方案、符号链接目标，以及 NTFS 备用数据流（冒号）向量，并将每个键解析到一个配置好的根目录之下。跨进程 worker 结果会被重新哈希，并与 worker 上报的摘要比对，因此一个被搅乱的 worker 无法悄悄损坏输出。本模块不记录任何文档内容。

Enterprise 边界说明

Stream 在此处的持久存储是文件系统支撑且单主机的。跨主机并发地向同一个键恰好一次写入，以及跨运行的持久去重，是 Enterprise 对象存储 committer 与存储的职责；驱动这些协作者的文档作业流处理器属于 Enterprise 范畴。Pro 提供引擎、契约，以及本地持久实现。

Core 回退/替代方案

没有 Pro 时，用 NextPDF Core 的 writer 一次渲染一份文档；持久批量流式、并发执行，以及恰好一次提交都是 Pro 的增项。参见 /modules/writer/。

发布边界

本页仅记录外部可观察的行为以及受支持的公共 API 表面。内部命名空间路径、辅助类、机制表、运行手册文件名，以及工单前缀均不在范围之内。

另请参阅

• Stream — 深度参考
• Stream（Enterprise）
• Writer
• Output Pipeline