LlosLab Software Development Blog面向 PDF、Delphi 和文档软件开发者的工程笔记

技术文章

Delphi 中的并发 ZIP 解压：HotXLS 读取闸门

HotXLS Component

面向 Delphi 和 C++Builder 的原生 Excel 组件库 HotXLS，能从一个打开的 ZIP 包里同时解压多个 XLSX 工作表。实现机制是 TZipReadGate，lxZipArchive.pas 里的一个小类，持有包流本身和一个临界区，只对外暴露一个方法。它串行化的是"定位加读取"这一对操作。这一对操作之上的一切都并发运行

逼出这个设计的问题，是每一个打开过大工作簿的 Delphi 开发者都遇到过的。一份 80 MB 的 xlsx 就是 80 MB 经过压缩的 XML，其中的工作表部件展开后大约是原来的五到十倍。如果打开路径在解析每张工作表之前，先把它整体解压进一个内存流，你就要在正在构建的工作簿之上，额外再付一份解压后字节的代价，而且这个内存峰值出现在任何一个单元格被创建出来之前。本文讲的正是移除了这个暂存步骤的包级并发机制。它之上的分配器上限在并行 XLSX 解析与内存管理器一文中有讲，只读一次、从不实体化的 API 在流式直读器详解一文中有讲

为什么旧的打开路径要把每张工作表都暂存进内存

HotXLS 原来的并行打开是一条三阶段流水线，中间那个阶段是唯一跑在工作线程上的阶段。阶段 A 在调用线程上串行遍历工作表列表，创建每张工作表，读取它的关系部件，并把解压后的整份工作表 XML 复制进一个私有的 TMemoryStream。阶段 B 把 ParseWorksheetXml 分发到线程池上。阶段 C 回到调用线程，再去归档里取那些小的附属部件：批注、线程化批注、绘图、图表、表格。这个形状是出于一个明确写出来的理由才选定的。lxParallelParse.pas 文件头部的注释原来明明白白地写着，ZIP 归档及其解压状态不是线程安全的，内部笔记说得更直接：不要费心去给归档加锁，因为一旦解压状态机按条目串行化，锁就什么都买不到了。阶段 A 存在的意义，就是把每一次归档访问都留在同一个线程上。代价是，一份有八张活跃工作表的工作簿，会同时在内存里持有八份完整解压出来的工作表 XML 缓冲区，而这些缓冲区正是整条打开路径上体积最大的临时对象

两个线程能不能从同一个 ZIP 流解压

能，而且旧的判断错在一个具体、可以定位的地方：它把两种不同的状态混进了同一句话里。解压状态确实不能共享。一个 zlib z_stream 承载着某一个压缩成员的滑动窗口、霍夫曼表和位位置，两个线程往同一个 z_stream 里塞字节，产出的只会是垃圾数据。而底层字节源完全是另一个问题，这里的答案是，一个文件流值得保护的可变共享状态其实只有一个：它的位置游标

ZIP 容器本身让这种拆分成为合法的做法。ZIP 归档里每个成员都是独立压缩的：各自有自己的本地文件头，各自在自己的 DataOffset 处有自己的 deflate 位流，中央目录里各自有自己的 CRC32 和大小。这里没有像 solid 7z 分块那样跨成员共享的字典，所以解压条目 N 完全不需要碰到条目 M。给每个工作线程各自的 z_stream，各自负责各自的字节范围，它们唯一会冲突的地方就是那次定位（seek）。TZipReadGate 消除的正是这个冲突，整个类短到一屏就能读完

type
  TZipReadGate = class
  private
    FBaseStream: TStream;
    FLock: TRTLCriticalSection;
  public
    constructor Create(ABaseStream: TStream);
    destructor Destroy; override;
    function ReadAt(AOffset: Int64; var Buffer; Count: Longint): Longint;
  end;

function TZipReadGate.ReadAt(AOffset: Int64; var Buffer;
  Count: Longint): Longint;
begin
  if Count <= 0 then
  begin
    Result := 0;
    Exit;
  end;
  EnterCriticalSection(FLock);
  try
    FBaseStream.Position := AOffset;
    Result := FBaseStream.Read(Buffer, Count);
  finally
    LeaveCriticalSection(FLock);
  end;
end;

TZipReadGate 保护的是什么，又刻意不保护什么

TZipReadGate.ReadAt 只守护一个不可分割的操作——给共享流定位并从中读取，别的什么都不管。TZipArchive.OpenArchive 在中央目录解析成功之后，会在 FInputStream 之上构造这道闸门，TZipArchive.Close 负责释放它。以写入方式打开的归档从来不会拿到一个这样的闸门。因此每个工作线程对这个包执行的每一次读取，都会汇聚经过同一个临界区，持有时长仅为一次带缓冲的读取操作

其余的一切都留在锁之外，因为它们要么本来就是私有的，要么本来就是不可变的。TZipSubStream 自己维护自己的 FPosition，所以每个工作线程在自己的条目里跟踪自己的位置。TZipEntry.GetStream 在这个子流之上构建的 TZLibStream，是每个条目各自独立的，用 windowBits 为 -15 创建以支持原始 deflate 格式，从不共享。中央目录在任何工作线程开始之前就已经完全解析完毕，包括每一个本地文件头，所以到并发开始的那一刻，GetEntryByName 已经是一次只读的哈希查找。路由本身只有 TZipSubStream.Read 里的三行代码，那条不经过闸门的分支，正是让所有既有的单线程调用方继续走老代码路径的原因

function TZipSubStream.Read(var buffer; Count: longint): longint;
var
  rest: Int64;
  rc: longint;
begin
  rest := FSize - FPosition;
  if (Count > rest) then
    Count := rest;
  if FReadGate <> nil then
    rc := FReadGate.ReadAt(FOffset + FPosition, buffer, Count)
  else
  begin
    FBaseStream.Position := FOffset + FPosition;
    rc := FBaseStream.Read(buffer, Count);
  end;
  FPosition := FPosition + rc;
  Result := rc;
end;

争用情况下，这道闸门的代价有多大

比"给整个归档加全局锁"这句话听起来的要小，原因在于 TZLibStream 恰好使用的那个粒度。它的输入缓冲区大小是 BufferSize，定义为 $4000，所以 ReadInputBuffer 每次补充都会拉取 16 KB 的压缩字节，交给 zng_inflate。因此，一次锁的获取覆盖的是 16 KB 的 deflate 输入，对工作表 XML 而言，这大约会展开成 100 KB 量级的标记文本，工作线程随后在不持有任何锁的情况下解码和解析这些内容。锁持有的时间只是一次针对操作系统缓存的定位读取；它所守护的工作量则是以毫秒计的

诚实地说，这个比例会在某处反转。存储而非压缩的条目，经过闸门的读取是一比一的，没有解压工作可以用来掩盖延迟，所以一个装满了存储型成员的包，串行化程度会重得多。一份放在慢速介质上的冷文件会拉长临界区的持有时间，因为里面的那次读取现在是一次真正的磁盘传输，而不是缓存命中。而且过了几个工作线程之后，闸门根本不是你首先撞上的瓶颈：工作表解析是分配密集型的，Delphi 内存管理器在跨线程的分配上，早在读取闸门成为瓶颈之前就已经开始串行化了。这也是为什么 TXLSXWorkbook.ParallelParseThreads 默认采用一个自动上限，而不是每核一个线程

工作线程主体，以及容易被遗忘的排空循环

有了这道闸门，HotXLS 干脆彻底删掉了阶段 A 的暂存步骤。工作线程现在直接打开自己的条目流，直接喂给解析器。两个临时字段负责传递输入：FParZip 在整个并行阶段持有归档，FParSheetPartNames 持有部件名称，两者都在 finally 块里清空，这样一次失败的打开就不会留下一个陈旧的指针。TZipArchive.OpenFile 返回的流是一个 TZipVerifiedStream，包着一个 TZLibStream，再包着一个 TZipSubStream，释放最外层就会释放整条链

procedure TXLSXWorkbook.ParseSheetJob(AIndex: Integer);
var
  Stream: TStream;
  DrainBuffer: array [0..32767] of Byte;
  PartName: WideString;
begin
  PartName := WideString(FParSheetPartNames[AIndex]);
  Stream := FParZip.OpenFile(PartName);
  if Stream = nil then
    Exit;
  try
    ParseWorksheetXml(Stream, FSheets.ByPos[AIndex], FParSst,
      FParRels[AIndex], FParFontMap, FParFillMap, FParBorderMap,
      FParNumFmtMap, FParAlignMap, FParProtMap, FParDateMap);
    // Consume any trailing bytes so the ZIP entry size and CRC are verified.
    while Stream.Read(DrainBuffer, SizeOf(DrainBuffer)) > 0 do
      ;
  finally
    Stream.Free;
  end;
end;

这个排空循环是那种直接照搬旧代码时容易被漏掉的细节，一旦悄悄漏掉，完整性校验就会被静默关闭。TZipVerifiedStream 在字节流经时累积一份运行中的 CRC32，只有当它的位置到达中央目录里记录的解压后大小时，才会调用 VerifyComplete；大小不匹配和 CRC32 不匹配这两个异常正是从这里抛出的，还有一次一字节的探测读取，用来捕获实际长度超出声明的条目。一个 XML 读取器通常在闭合标签处就停下了，往往还会留下一个换行符或几个字节的尾部空白没有读取，所以如果不做排空，位置就永远到不了声明的大小，这些检查也就永远不会触发。把剩余部分读进一个暂存缓冲区几乎不花什么成本，却能让这些检查重新生效。在还存在暂存流的年代，XlsxCopyStreamAll 其实是无意中顺带做了这件事

依然串行运行的部分，以及把这一切整体关掉的开关

阶段 A 保留了下来，只是去掉了提取那一步。它依然在调用线程上创建每张工作表、读取它的关系，这也正是让每一份共享映射表在工作线程开始之后保持不可变的原因。阶段 C 依然在之后串行遍历各张工作表，处理批注、绘图、图表和表格，它的判断条件从对旧暂存数组做空值检查，改成了对照部件名称做 zip.Exists。工作线程会碰到的那些共享只读输入——共享字符串表和 cellXf 映射表——都在阶段 B 开始之前就已经完备，在阶段 B 期间从不会被写入

var
  Wb: TXLSXWorkbook;
begin
  Wb := TXLSXWorkbook.Create;
  try
    Wb.ParallelParse := True;      // default; False forces one sheet at a time
    Wb.ParallelParseThreads := 4;  // 0 selects the automatic cap
    Wb.Open('quarterly-consolidation.xlsx');
    // ... workbook is identical either way ...
  finally
    Wb.Free;
  end;
end;

在 Open 之前把 ParallelParse 设为 False，会以线程数为一去分发同一个作业过程，RunParallelJobs 会退化成调用线程上的一个普通循环。这一点值得知道，原因有两个：如果线程相关的问题在现场出现了，这是最简单的一行答案；而且这意味着串行路径和并行路径共享同一份解析代码，而不是各自分叉。工作线程的异常会被捕获，索引最小的那个作业胜出，错误会在所有工作线程都汇合之后，在调用线程上重新抛出，所以一张损坏的工作表依然会在预期的地方以一个异常的形式浮现出来。围绕这条打开路径的通用调优在Delphi 大型工作簿性能指南一文中有讲

这里描述的读取闸门、并行打开阶段和流式条目访问，都作为标准版 HotXLS Excel 组件（面向 Delphi 和 C++Builder，带完整源码）的一部分随附提供；产品页带有 TXLSXWorkbook 的完整参考，包括并行打开相关的属性