PDF.js 从 demo 深入浅出 7: WorkerTransport 初始化

前言

第 6 篇解释了 MessageHandler 如何传输消息, 但它本身不理解 GetDoc、GetReader 这些名字的业务含义. 负责解释这些消息的是本篇的主角: WorkerTransport.

回到第 2 篇的握手末尾, worker 对 GetDocRequest 返回 workerId 后, 主线程执行:

const messageHandler = new MessageHandler(docId, workerId, worker.port);
const transport = new WorkerTransport(
  messageHandler,
  task,
  networkStream,
  transportParams,
  transportFactory,
  pagesMapper
);
task._transport = transport;
 
if (task.destroyed) {
  throw new Error("Loading aborted");
}
messageHandler.send("Ready", null);

本篇从 new WorkerTransport(...) 开始, 一直走到第一篇 demo 的:

const loadingTask = pdfjsLib.getDocument(url);
const pdf = await loadingTask.promise;

回答三个问题:

1. WorkerTransport 构造时准备了什么?
2. Ready 发出后, PDF 字节如何开始流向 worker?
3. loadingTask.promise 究竟在哪一刻、由谁 resolve?

本篇只跟随 demo 的文档加载主干, 不逐个展开 WorkerTransport 的所有 API. 完整的字段与消息分类可对照 api.js 源码解读.

WorkerTransport 调用

依据 src/display/api.js 中 WorkerTransport 的定义, 初始化过程可以分成「保存依赖」「注册消息」「等待 Ready」三步.

第 1 步, 六个构造参数各就各位

demo 只给 getDocument 传了 url, 但前面的参数解析与握手已经为主线程准备好了六个依赖:

这里的 workerId 是 createDocumentHandler 创建的文档专属 worker 名称. demo 第一份文档中 docId = "d1", 因而 workerId = "d1_worker".

构造函数先保存依赖并创建主线程侧资源容器:

this.messageHandler = messageHandler;
this.loadingTask = loadingTask;
this.#networkStream = networkStream;
 
this.commonObjs = new PDFObjects();
this.fontLoader = new FontLoader({
  ownerDocument: params.ownerDocument,
  styleElement: params.styleElement,
});
this.enableHWA = params.enableHWA;
this.loadingParams = params.loadingParams;
this._params = params;
 
this.canvasFactory = factory.canvasFactory;
this.filterFactory = factory.filterFactory;
this.binaryDataFactory = factory.binaryDataFactory;
this.pagesMapper = pagesMapper;
 
this.destroyed = false;
this.destroyCapability = null;

此外还有三个页面缓存与一个方法缓存, 在类字段声明时就已经创建:

#methodPromises = new Map();
#pageCache = new Map();
#pagePromises = new Map();
#pageRefCache = new Map();

此时它们全是空 Map. 第 3 篇 getPage(1) 使用的 #pagePromises、#pageCache、#pageRefCache, 就是在这里准备好的.

注意: 构造函数没有读取 PDF, 也没有调用 networkStream.getFullReader(). 它只是把后续加载、渲染、销毁所需的依赖与容器装配好.

第 2 步, setupMessageHandler 注册业务路由

字段赋值结束后, 构造函数调用:

this.setupMessageHandler();

它通过一组 messageHandler.on(name, handler) 注册主线程侧业务处理器. 按 demo 中的执行时机, 可以这样分组:

注册不等于执行. demo 当前只关心其中的数据供给与文档结果消息; StartRenderPage 要等第 5 篇的 render 才会用到.

设计要点: 先注册全部 handler, 再发送 Ready. 这样 worker 一旦开始加载, 无论首先发来的是 GetReader、PasswordRequest 还是 DocException, 主线程都已经有对应的接收方. 顺序反过来就会出现「消息到了, 路由表还没准备好」的竞态.

第 3 步, 挂到 task 后再发送 Ready

WorkerTransport 返回后, getDocument 先执行 task._transport = transport, 再检查一次 task.destroyed, 最后才发送:

messageHandler.send("Ready", null);

这次检查处理的是握手期间的销毁竞态: 用户可能已经在异步等待中调用了 loadingTask.destroy(). 如果文档已销毁, 就不应该再让 worker 开始加载.

demo 没有调用 destroy, 所以 Ready 正常发出. 它是第 6 篇所说的纯 action: 不需要回包, 只负责通知 worker「主线程的消息路由与数据供给都已就绪」.

Ready 之后: worker 开始索取 PDF

第 4 步, Ready 触发 setupDoc

worker 在先前处理 GetDocRequest 时, 已经创建文档专属 MessageHandler 并注册了 Ready 处理器, 但尚未开始解析:

handler.on("Ready", function () {
  setupDoc(docParams);
  docParams = null; // 后续不再需要, 释放参数包
});

收到主线程的 Ready 后, setupDoc(docParams) 才启动. demo 的 docParams.data = null, 因此 worker 不能直接拿到 PDF 字节, 必须建立一条反向数据通道:

worker 解析器需要字节
  → 通过消息向主线程索取
  → 主线程从 networkStream 读取
  → 通过消息把 ArrayBuffer 转交 worker

这就是为什么网络流建在主线程, 而解析器运行在 worker.

第 5 步, PDFWorkerStream 发出 GetReader

setupDoc 调用 getPdfManager(data). 在 data = null 的 demo 分支中, worker 创建自己的流对象:

const pdfStream = new PDFWorkerStream({ msgHandler: handler });
const fullReader = pdfStream.getFullReader();

PDFWorkerStream 位于 src/core/worker_stream.js. 它不访问网络, 只把 reader 操作翻译成消息. 全量 reader 构造时立即做两件事:

const readableStream = msgHandler.sendWithStream("GetReader");
this._reader = readableStream.getReader();
 
msgHandler.sendWithPromise("ReaderHeadersReady").then(data => {
  this._contentLength = data.contentLength;
  this._isStreamingSupported = data.isStreamingSupported;
  this._isRangeSupported = data.isRangeSupported;
  this._headersCapability.resolve();
}, this._headersCapability.reject);

两个消息分工不同:

• GetReader: 建立长期的流式字节通道;
• ReaderHeadersReady: 单次请求响应, 获取长度与加载能力.

这与第 6 篇的三种消息形态对应起来: 前者是 sendWithStream, 后者是 sendWithPromise.

第 6 步, 主线程创建真正的网络 reader

主线程的 GetReader 处理器开始执行:

messageHandler.on("GetReader", (data, sink) => {
  this.#fullReader = this.#networkStream.getFullReader();
  this.#fullReader.onProgress = evt => this.#onProgress(evt);
 
  sink.onPull = () => {
    this.#fullReader.read().then(({ value, done }) => {
      if (done) {
        sink.close();
        return;
      }
      sink.enqueue(new Uint8Array(value), 1, [value]);
    }).catch(reason => {
      sink.error(reason);
    });
  };
 
  sink.onCancel = reason => {
    this.#fullReader.cancel(reason);
    // 后续 sink.ready 的清理略
  };
});

直到调用 networkStream.getFullReader(), demo 的 PDFFetchStreamReader 才被创建并发起 fetch(url). 也就是说:

getDocument({ url }) 创建网络流配置, new WorkerTransport(...) 注册数据处理器, Ready 让 worker 发出 GetReader——到这里才真正开始 HTTP 请求.

ReaderHeadersReady 则等待这次请求拿到响应头:

messageHandler.on("ReaderHeadersReady", async data => {
  await this.#fullReader.headersReady;
 
  const { isStreamingSupported, isRangeSupported, contentLength } =
    this.#fullReader;
 
  if (isStreamingSupported && isRangeSupported) {
    this.#fullReader.onProgress = null;
  }
  return { isStreamingSupported, isRangeSupported, contentLength };
});

worker 根据这三个值决定后续策略:

• 支持 range: 建立 NetworkPdfManager, 后续可按需发送 GetRangeReader;
• 不支持 range: 持续读取全量流, 数据齐全后建立 LocalPdfManager.

demo 的 rangeChunkSize = 65536、disableRange = false、disableStream = false 会参与这里的判断, 但能否使用 range 还取决于响应头. 具体探测与请求策略见第 9 篇.

第 7 步, onPull 形成按需字节泵

GetReader 处理器里的 sink.onPull 不会在注册时主动执行, 而是在 worker 的 ReadableStream 需要更多数据时被触发:

worker fullReader.read()
  → MessageHandler 发出 PULL
  → 主线程 sink.onPull()
  → PDFFetchStreamReader.read()
  → fetch ReadableStream 返回一块字节
  → sink.enqueue(new Uint8Array(value), 1, [value])
  → worker 收到 ENQUEUE

sink.enqueue 的第三个参数 [value] 将底层 ArrayBuffer 列为 Transferable: buffer 的所有权转给 worker, 避免复制整段 PDF 数据.

因此这不是主线程一次性把整个文件推给 worker, 而是worker 按消费能力逐块拉动. 当 done = true 时主线程调用 sink.close(), 字节通道结束.

GetDoc: loadingTask.promise 的完成点

第 8 步, worker 解析文档级元数据

数据通道建立后, setupDoc 等待 getPdfManager 给出 pdfManager, 随后进入 pdfManagerReady → loadDocument(false).

loadDocument 与本文主干相关的部分是:

async function loadDocument(recoveryMode) {
  await pdfManager.initDocument(recoveryMode);
 
  // XFA 分支略
  const [numPages, fingerprints] = await Promise.all([
    pdfManager.ensureDoc("numPages"),
    pdfManager.ensureDoc("fingerprints"),
  ]);
  const htmlForXfa = isPureXfa
    ? await pdfManager.ensureDoc("htmlForXfa")
    : null;
 
  return { numPages, fingerprints, htmlForXfa };
}

成功后, worker 发送:

handler.send("GetDoc", {
  pdfInfo: { numPages, fingerprints, htmlForXfa },
});

demo 文档的 numPages = 756; 普通非 XFA 文档的 htmlForXfa = null. fingerprints 是文档标识数组.

这里发送的是 GetDoc, 不是 GetDocRequest 的 callback:

• GetDocRequest 的回执早已返回, 内容只有 workerId, 用来建立文档专属通道;
• GetDoc 是 worker 之后主动发送的纯 action, 内容是文档元数据, 用来宣告「文档已经可以操作」.

第 9 步, 主线程创建 PDFDocumentProxy

主线程先前注册的 GetDoc 处理器收到 pdfInfo:

messageHandler.on("GetDoc", ({ pdfInfo }) => {
  this.pagesMapper.pagesNumber = pdfInfo.numPages;
  this._numPages = pdfInfo.numPages;
  this._htmlForXfa = pdfInfo.htmlForXfa;
  delete pdfInfo.htmlForXfa;
  loadingTask._capability.resolve(new PDFDocumentProxy(pdfInfo, this));
});

它依次完成四件事:

1. 把 pagesMapper.pagesNumber 从初始值 0 更新为 756, 第 3 篇 getPage(1) 的页码上限校验由此获得依据;
2. 在 transport 上保存 _numPages 与 _htmlForXfa, 供后续保存文档、XFA 渲染等流程使用;
3. 从公开的 pdfInfo 删除 htmlForXfa, XFA HTML 树只留在 transport 内部;
4. 创建 PDFDocumentProxy, 并用它 resolve loadingTask._capability.

PDFDocumentProxy 的构造函数非常轻:

constructor(pdfInfo, transport) {
  this._pdfInfo = pdfInfo;
  this._transport = transport;
}

它不解析 PDF, 只保存已经到达的文档信息与 transport 引用. 因此读取总页数不需要再发消息:

get numPages() {
  return this._pdfInfo.numPages; // 756
}

而 pdf.getPage(1) 才会通过 this._transport 发起新的 GetPage 请求.

第 10 步, demo 的 await 恢复执行

loadingTask.promise 是 _capability.promise 的 getter. 上一步调用 _capability.resolve(pdfDocumentProxy) 后, 第一篇 demo 中的等待结束:

const pdf = await loadingTask.promise;
// pdf 是 PDFDocumentProxy
// pdf.numPages === 756
// pdf._transport === 本篇创建的 WorkerTransport

随后 demo 才执行 pdf.getPage(1), 进入第 3 篇的调用链.

文档就绪不等于下载完成

GetDoc 与 DataLoaded 是两条独立消息:

支持 range 时, worker 可能只读取 xref、对象字典等必要区域就能发送 GetDoc, 其余页面内容仍留在服务器上. 所以 await loadingTask.promise 结束后, 不代表整份 PDF 已下载, 也不代表第一页已解析或渲染.

DataLoaded 到达时, 主线程只是补一次完成进度并记录下载信息:

messageHandler.on("DataLoaded", data => {
  this.#onProgress({ loaded: data.length, total: data.length });
  this.downloadInfoCapability.resolve(data);
});

demo 没有注册 loadingTask.onProgress, 因而进度回调不会产生外部可见效果.

如果加载失败, DocException 会 reject 与 GetDoc 相同的 capability:

messageHandler.on("DocException", ex => {
  loadingTask._capability.reject(wrapReason(ex));
});

所以 demo 的 await loadingTask.promise 只有两种结果: GetDoc 带来 PDFDocumentProxy, 或 DocException 带来异常. 加密文档还会先经过 PasswordRequest; demo 没有设置 onPassword, 无法提供密码时最终同样进入失败路径.

总结

本篇跟随 demo 从 new WorkerTransport(...) 走到了 await loadingTask.promise 完成:

1. 构造函数保存 MessageHandler、loadingTask、networkStream、三个工厂与 pagesMapper, 创建资源容器和空缓存;
2. setupMessageHandler() 注册数据供给、文档结果、进度、异常与后续渲染的消息处理器;
3. transport 挂到 task._transport, 销毁检查通过后, 主线程发送 Ready;
4. worker 收到 Ready 才执行 setupDoc, 创建 PDFWorkerStream;
5. GetReader 建立流式数据通道, ReaderHeadersReady 返回 range/streaming/contentLength 三项能力;
6. 主线程首次调用 networkStream.getFullReader(), 真正发起 fetch, 并通过 sink.onPull 按 worker 需求逐块送回字节;
7. worker 初始化 pdfManager, 读取 numPages 与 fingerprints, 主动发送 GetDoc;
8. 主线程更新总页数, 创建 PDFDocumentProxy, resolve loadingTask._capability;
9. demo 的 await 恢复, 得到 pdf, 随后才能进入 getPage(1).

WorkerTransport 因而不只是消息转发器: 它同时持有网络流、文档状态、页面缓存、字体与 canvas 工厂, 是主线程的文档运行时与业务语义层. MessageHandler 负责「消息怎么到」, WorkerTransport 负责「到了以后做什么」.

第 8 篇先回到第 2 篇的 PDFWorker.create 钩子, 展开 worker 创建、ready/test 握手与假 worker 回退. 随后第 9 篇再沿本篇的 GetReader 向下, 展开网络流的 range 能力探测与 GetRangeReader 请求策略. 也可先对照网络流源码解读查看完整的四种实现.

StreamPDF 的设计原则: 能前置处理的绝对不放到解析的时候才做, 由服务端一次解决所有性能问题. 浏览器只获取&渲染必须数据.