PDF.js 从 demo 深入浅出 6: MessageHandler 跨线程通讯

前言

第 2 篇的总结列了四个「重角色」, 本篇处理第二个: MessageHandler 的消息机制.

前面五篇 demo 里, MessageHandler 出现过四次, 每次调用形态不同:

前两篇形态都是「单次请求, 单次答复」, 第四篇是「生产端分块推, 消费端按背压拉」的流式通道. 本篇从第 2 篇的创建处入手, 顺着这四次调用, 把 MessageHandler 内部的信封协议和三种形态的完整链路走一遍.

如果你想同时看调用方视角 (api.js 文章里的用法) 与协议实现视角 (本文), 请对照 MessageHandler 源码解读. 本篇不重复源码解读里的 API 列表, 只聚焦 demo 中实际发生的执行路径.

创建: 握手拿到 workerId 之后

第 2 篇 第 261 行 worker 回了 GetDocRequest 的回执 workerId, 主线程立刻建一个 MessageHandler:

const messageHandler = new MessageHandler(docId, workerId, worker.port);
// docId = "d1"
// workerId = worker 端自分配的 id (demo 里通常也是 "worker" 或类似字符串)
// worker.port = Worker 对象本身 (Worker 也是一个 MessagePort-like)

三个参数分别填进实例字段:

this.sourceName = "d1";       // 我是谁
this.targetName = workerId;   // 我跟谁说话
this.comObj     = worker.port; // 发消息走哪
this.callbackId = 1;          // 下次请求的 callbackId
this.streamId   = 1;          // 下次流式通道的 streamId
this.callbackCapabilities = {}; // callbackId → Promise.withResolvers()
this.actionHandler        = {}; // actionName → handler (WorkerTransport 后续注册)
this.streamControllers    = {}; // streamId → ReadableStream controller (我发起的流)
this.streamSinks          = {}; // streamId → sink (对端发起的流, 我是生产端)
 
comObj.addEventListener("message", this.#onMessage.bind(this), { signal: this.#messageAC.signal });

构造函数只做两件事: 建四个空 Map, 把 #onMessage 绑进 worker.port 的 message 事件. 此时 actionHandler 还是空的——所有业务 handler 由后续 WorkerTransport.setupMessageHandler() 通过 .on(...) 注册.

设计要点: MessageHandler 本身完全不知道业务语义. 它只认"信封"的三个关键字段 (action, callbackId, streamId), 不认 GetDocRequest / GetPage 这些 action 名. 业务语义全在 WorkerTransport 的 .on(...) 注册表里. 这让 MessageHandler 可以被主线程↔worker、主线程↔主线程 (LoopbackPort)、甚至未来主线程↔iframe 等任意两端复用.

信封协议: 三形态互斥分发

所有穿过 comObj.postMessage 的消息都套同一个信封. 信封之上再按是否带 stream / callback / callbackId / streamId 四个关键字段分形态:

#onMessage({ data }) {
  if (data.targetName !== this.sourceName) return;
 
  if (data.stream)   { /* 流控制或流数据: PULL/ENQUEUE/CLOSE/... */ }
  if (data.callback) { /* 纯回包: 对端给我答复, 带 callbackId */ }
 
  // 剩下的才是 action
  const action = this.actionHandler[data.action];
  if (data.callbackId) { /* action + callbackId: 执行后必须回 callback */ }
  else if (data.streamId) { /* action + streamId: 执行后建 sink, 不回 action 形态答复 */ }
  else { /* 纯 action, fire-and-forget */ }
}

严格顺序: stream → callback → action, 互斥, 一次消息只走一条路. 两个字段同时出现 (比如既带 stream 又带 callback) 说明信封被错误组装, #onMessage 会漏掉其中一个.

sourceName / targetName 是软路由: 匹配不到就 return, 不抛错. demo 只有一个文档 (d1 ↔ worker), 所有消息都匹配; 如果同时跑两份文档, 第二份 d2 的 MessageHandler 会自动忽略 d1 的消息, 无需额外隔离逻辑.

形态一: 纯 action —— Ready 宣告

第 2 篇 握手最后一步:

messageHandler.send("Ready", null);

send 是最简单的一条通道:

send(actionName, data, transfers) {
  this.comObj.postMessage({
    sourceName: this.sourceName,  // "d1"
    targetName: this.targetName,  // "worker"
    action: actionName,           // "Ready"
    data,                         // null
  }, transfers);                  // undefined
}

信封里没有 callbackId, 没有 streamId, 没有 stream 字段. 对端 worker 的 MessageHandler 收到后, #onMessage 走到最后一条 else, 直接调 this.actionHandler["Ready"]. worker 端注册了这个 handler, 内部做"把自己标记为已就绪"之类的事. 做完就完了——不回包. 这就是 fire-and-forget.

demo 里这条消息很容易被忽略, 因为它不影响任何 promise 或返回值. 但 WorkerTransport 在 setupMessageHandler 里注册了大量带 callbackId 的 action handler (GetDoc / GetPage / SaveDocument 等), 所有请求-应答通道都在那里. send 只用于"通知型"消息: Ready、configure(verbosity)、以及后续的 Terminate.

形态二: action + callbackId —— GetPage 取第一页

demo 最典型的请求-应答发生在第 3 篇的 pdf.getPage(1):

const promise = messageHandler
  .sendWithPromise("GetPage", { pageIndex: 0 });

sendWithPromise 的发送侧

sendWithPromise(actionName, data, transfers) {
  const callbackId = this.callbackId++;  // 1, 每次请求递增
  const capability = Promise.withResolvers();
  this.callbackCapabilities[callbackId] = capability;
  try {
    comObj.postMessage({
      sourceName: "d1", targetName: "worker",
      action: "GetPage", callbackId: 1,
      data: { pageIndex: 0 },
    }, transfers);
  } catch (ex) {
    capability.reject(ex);
  }
  return capability.promise;
}

与 send 唯一区别: 带了 callbackId, 并把 capability resolver 记在 this.callbackCapabilities[1] 表里. postMessage 返回后这条 promise 还在 pending —— 必须等 worker 端回一条 callback 形态的消息来 resolve.

信封里没有 streamId, 所以对端 worker 的 #onMessage 不会走流式分支. 它走到 action 分支, 发现信封带 callbackId, 就按"执行后回包"处理.

对端处理: action + callbackId 的语义

worker 端 MessageHandler 的 #onMessage 拿到信封:

const action = this.actionHandler["GetPage"]; // worker 端注册的 handler
if (data.callbackId) {
  Promise.try(action, data.data).then(
    result => comObj.postMessage({
      sourceName: "worker", targetName: "d1",
      callback: CallbackKind.DATA,
      callbackId: 1,
      data: result, // 解析好的 pageInfo
    }),
    reason => comObj.postMessage({
      sourceName: "worker", targetName: "d1",
      callback: CallbackKind.ERROR,
      callbackId: 1,
      reason: wrapReason(reason),
    })
  );
  return;
}

信封的 action 字段在回包里不再出现——回包是纯 callback 形态, 只带 callback (DATA=1 或 ERROR=2), callbackId, data 或 reason.

回到主线程: callback 分支 resolve 悬挂的 promise

主线程 #onMessage 收到这条回包, 走 callback 分支:

if (data.callback) {
  const callbackId = data.callbackId; // 1
  const capability = this.callbackCapabilities[callbackId]; // 就是之前 sendWithPromise 存的那个
  delete this.callbackCapabilities[callbackId];
 
  if (data.callback === CallbackKind.DATA) {
    capability.resolve(data.data); // promise resolve 出 pageInfo
  } else if (data.callback === CallbackKind.ERROR) {
    capability.reject(wrapReason(data.reason));
  }
}

Promise 回主线程事件循环, demo 里 await promise 的后续代码拿到 pageInfo, 构造 PDFPageProxy 返回.

sendWithPromise 的同步 try-catch 兜什么

try {
  comObj.postMessage({ /* ... */ }, transfers);
} catch (ex) {
  capability.reject(ex);
}

只兜 postMessage 自身抛错: 例如 transfer 里传了已被 transfer 所有权的 ArrayBuffer, 或 data 里有不可序列化的对象. postMessage 返回后, worker 端 handler 执行期间的任何异常都已经在另一个事件循环, 必须走 callback ERROR 通道才能回来——try-catch 管不到.

这就是为什么 this.callbackCapabilities[callbackId] = capability 必须先存再 postMessage: 如果 postMessage 抛错, catch 块里能直接找到 capability reject 掉, 不留悬挂.

形态三: action + streamId —— GetOperatorList 流式渲染

demo 里最复杂的通道出现在第 5 篇 page.render(...) 触发的 _pumpOperatorList:

const readableStream = messageHandler.sendWithStream(
  "GetOperatorList",
  { pageId: 0, pageIndex: 0, intent: 2, cacheKey: "2__", ... },
  undefined,  // queueingStrategy
  transfer
);
const reader = readableStream.getReader();
reader.read().then(({ value, done }) => { /* 追加进 operatorList */ });

sendWithStream: ReadableStream start 回调作为 "建通道请求"

sendWithStream(actionName, data, queueingStrategy, transfers) {
  const streamId = this.streamId++; // 1
  return new ReadableStream({
    start: controller => {
      const startCapability = Promise.withResolvers();
      this.streamControllers[streamId] = {
        controller, startCall: startCapability,
        pullCall: null, cancelCall: null, isClosed: false,
      };
      comObj.postMessage({
        sourceName: "d1", targetName: "worker",
        action: "GetOperatorList", streamId: 1,
        data: { pageId: 0, pageIndex: 0, intent: 2, cacheKey: "2__" },
        desiredSize: controller.desiredSize, // 初始值, ReadableStream 告诉你"我还能接多少"
      }, transfers);
      return startCapability.promise;
    },
    // pull / cancel 见下文
  }, queueingStrategy);
}

关键: start 回调返回一个 promise. ReadableStream 会等这个 promise resolve 后, 才开始向消费端推数据. 所以主线程在 start 里先把 streamId=1 + action 发给 worker, 再把 startCapability.promise 返回给 ReadableStream. Worker 回 START_COMPLETE 后这个 promise resolve, ReadableStream 才开始接受 pull 请求.

信封带 streamId 但不带 stream 字段——它是 action + streamId 形态, 不是流控制形态.

对端: action + streamId → #createStreamSink

worker 端 MessageHandler 的 #onMessage 看到 data.streamId = 1, 走 this.#createStreamSink(data):

// 为 streamId=1 创建 sink 对象:
const streamSink = {
  enqueue(chunk, size = 1, transfers) { /* 发 ENQUEUE */ },
  close()                             { /* 发 CLOSE   */ },
  error(reason)                       { /* 发 ERROR   */ },
  sinkCapability: Promise.withResolvers(),
  onPull:   null, // 主线程 pull 到来时回调我
  onCancel: null, // 主线程 cancel 到来时回调我
  isCancelled: false,
  desiredSize: data.desiredSize, // 从 start 消息里拿的初始值
  ready: null,   // 当 desiredSize <= 0 时挂起, pull 到来时 resolve
};
this.streamSinks[1] = streamSink;
 
Promise.try(action, data.data, streamSink).then(
  () => comObj.postMessage({ stream: START_COMPLETE, streamId: 1, success: true }),
  reason => comObj.postMessage({ stream: START_COMPLETE, streamId: 1, reason: wrapReason(reason) })
);

sink 是主动生产者: worker 端的 GetOperatorList handler 拿到 sink 后, 每解析完一块绘制指令就调 sink.enqueue(chunk) 把数据推给主线程; 解析完所有块时 sink.close(). 对应的 onPull / onCancel 是被动响应——当主线程 ReadableStream 需要更多数据或被取消时, MessageHandler 会回调这两个函数.

ready 字段是 sink 侧的背压信号: sink.desiredSize 是主线程 ReadableStream 告诉 sink "还能接多少", 每 enqueue 一块就扣减; 扣到 ≤ 0 时 ready 被挂起, worker 侧应该在每次 enqueue 前 await sink.ready 避免无脑推送撑爆对方缓冲区.

主线程: START_COMPLETE → startCapability resolve

worker 回 START_COMPLETE, 主线程 #processStreamMessage 处理:

case StreamKind.START_COMPLETE:
  if (data.success) {
    streamController.startCall.resolve();
  } else {
    streamController.startCall.reject(wrapReason(data.reason));
  }
  break;

startCapability resolve → ReadableStream 内部认为 start 已完成 → 开始向消费端推数据. demo 里 reader.read() 第一次被调用时, ReadableStream 内部 buffer 为空 → ReadableStream 触发自己的 pull 回调 → 发送 PULL 消息.

pull 链: 主线程拉 → worker 推 → 回 ENQUEUE → ReadableStream controller.enqueue

主线程 ReadableStream controller 需要更多数据
  → trigger pull() 回调
  → sendWithStream 里 pull 回调发 PULL 消息给 worker
  → worker #processStreamMessage 处理 PULL
    → reset sink.desiredSize = data.desiredSize
    → 调 sink.onPull  (或默认空函数)
    → 回 PULL_COMPLETE
  → 主线程收到 PULL_COMPLETE
    → streamController.pullCall.resolve
    → ReadableStream 开始从 worker 拉数据
  → worker GetOperatorList handler 解析完一块
    → sink.enqueue(chunk)
    → 发 ENQUEUE 消息
  → 主线程 #processStreamMessage 处理 ENQUEUE
    → streamController.controller.enqueue(data.chunk)

每轮 pull → 若干次 enqueue → close. ReadableStream 内部的 desiredSize 会在消费端读取时自动更新, 下次 pull 就带新值.

取消链: 主线程 cancel → worker CANCEL → CANCEL_COMPLETE

demo 里第 5 篇提到的「100ms 延迟取消」最终会调 readableStream.cancel(reason):

demo: streamReader.cancel(renderingCancelledException)
  → ReadableStream trigger cancel() 回调
  → sendWithStream 里 cancel 回调发 CANCEL 消息
  → worker #processStreamMessage 处理 CANCEL
    → 调 sink.onCancel(reason)
    → 回 CANCEL_COMPLETE
    → sink.isCancelled = true
    → sink.sinkCapability.reject(reason)
  → 主线程收到 CANCEL_COMPLETE
    → streamController.cancelCall.resolve
    → #deleteStreamController 等三条 promise settle 后清理

sink 侧也必须同步清理: sink.isCancelled = true 让后续的 sink.enqueue / sink.close / sink.error 变成空操作——防止 worker 继续往一个已取消的通道里灌数据.

8 种 StreamKind 的位置一览

三种形态的 demo 对照表

三条通道互不干扰: 同一个 comObj (worker.port) 同时跑三条消息流, 靠 callbackId / streamId 字段路由到正确的表项. 任何一条通道的消息, 信封里的 sourceName / targetName 都精确到文档 (d1 ↔ worker), 多文档并存时自动隔离.

回钩: WorkerTransport 是真正的"语义层"

第 2 篇里 MessageHandler 创建之后, 紧接着 WorkerTransport 构造函数调了 messageHandler.on("GetDoc", handler)、.on("DocException", ...) 等 20 多次注册——这些 handler 才是 MessageHandler 另一端要跑的"业务". MessageHandler 自己不知道 GetDoc 和 GetPage 有什么不同, 它只知道"信封上写了 action, 就去 actionHandler 里找".

这层分工意味着: 任何自定义 action 都可以插进来, 只要两端都注册了 handler. 除了 pdf.js 自带的 20+ 种, 用户自定义的扩展 (editor 里的 Stamp / FreeText 等) 也走同一套通道——MessageHandler 不区分内置 action 与扩展 action.

总结

本篇跟随 demo 中三次具体调用, 把 MessageHandler 内部的完整链路走了一遍:

1. 信封协议: 四关键字段 (action / callbackId / streamId / stream) 决定消息走哪条路, sourceName/targetName 做软路由;
2. 纯 action (send): Ready 宣告, 不回包;
3. action + callbackId (sendWithPromise): GetPage 请求, worker 执行后回 callback (DATA 或 ERROR), 主线程从 callbackCapabilities 表 resolve 悬挂的 promise;
4. action + streamId (sendWithStream): GetOperatorList 流式通道, ReadableStream start/pull/cancel 三回调与对端 sink 配对, 8 种 StreamKind 组成双向背压状态机.

三条通道互不干扰, 同一个 comObj 同时承载, 靠 callbackId/streamId 精确路由. MessageHandler 本身无业务语义, 所有 action 的语义全在 WorkerTransport 的 .on(...) 注册表里——它是最小原语封装.

想同时看源码层 API 列表与实现细节, 请对照 MessageHandler 源码解读. 想走下一个 demo 钩子? 回到第 2 篇的四个重角色清单, 接下来是 WorkerTransport 初始化 (demo 第 2 篇第 4 步紧接着 MessageHandler 创建之后的那块).

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