Reactor 将文件并行解压缩到内存导致 OutOfMemoryError答案

【问题标题】：Reactor unzip files in parallel to memory leads to OutOfMemoryErrorReactor 将文件并行解压缩到内存导致 OutOfMemoryError
【发布时间】：2021-08-05 13:46:21
【问题描述】：

我正在尝试处理一个包含多个要处理的内部 XML 文件的巨大 tar 文件。我有磁盘限制，所以我必须使用 uncompress-in-memory-do-stuff-then-forget 方法。我正在使用Apache Compress 库。到目前为止一切顺利，一切正常，但我现在想并行处理文件以缩短执行时间。为此，我们使用了反应器。

这似乎一直在工作，直到由于某种原因，数据在处理后没有被清理，所以它最终出现java.lang.OutOfMemoryError: Java heap space 异常。我考虑过的其他事情是调整垃圾收集器以更快地介入，但我有点迷茫，所以我很感激任何帮助。

进入代码。这是我正在使用的Flux.generate。对于每个下一个请求的项目，它会生成一个字节数组，其中包含来自 tar 文件的一个未压缩文件。看起来像这样：

(ArchiveInputStream tarArchive, SynchronousSink<byte[]> sink) -> {
    process: try {
      ArchiveEntry entry = null;
      while ((entry = tarArchive.getNextEntry()) != null) {
        if (entry.isDirectory()) continue;
        log.info(entry.getName());

        var bytes = IOUtils.readRange(tarArchive, (int) entry.getSize());
        sink.next(bytes);

        break process;
      }
      log.info("done!!!!");
      sink.complete();
    } catch (IOException e) {
      log.error("Exception getting file entry", e);
      sink.error(e);
    }
    return tarArchive;
  }

稍后，字节数组被转换为 ByteArrayInputStream 被 JAXB 映射到内部模型中：

  public Feed processElement(byte[] file) {
    var feed = new Feed();
    if (file.length != 0) {
      try(var is = new ByteArrayInputStream(file)) {
        var jaxbUnmarshaller = jaxbContext.createUnmarshaller();
        feed = (Feed) jaxbUnmarshaller.unmarshal(is);
      } catch (Exception ex) {
        log.error("Error during XML unmarshal", ex);
      }
    }
    return feed;
  }

流并行化如下：

uncompressFiles(tarFileName)
        .parallel()
        .runOn(parallel())
        .map(parser::processElement)
        .sequential()
//.....

如果您对这种特殊情况或如何解决内存泄漏有任何见解，请告诉我。

【问题讨论】：

调整垃圾收集器不太可能解决您的问题（假设错误消息“Java 堆空间”，而不是“超出 GC 开销限制”），因为 Java 将在抛出之前执行完整的垃圾收集那个错误。
是的，你是对的，它是 java.lang.OutOfMemoryError: Java heap space。我会更新de描述。我最初的猜测是引用挂在某处（可能与文件字节数组有关），但我无法找到它，所以我开始寻找其他可能性。
您确定引用被保留吗？您应该能够通过堆转储来确认这一点。这也可以通过并行处理将使用更多工作内存这一事实来解释。有多少并行进程正在运行？
parallel() 默认使用可用内核的数量，但您可以调整它。我尝试了几种工作线程和 JVM 最大堆配置，最终结果是相同的。虽然，我注意到删除并行进程可以保持内存稳定，但是将并行处理设置为 1 并没有，所以问题可能与将字节数组从主线程中从 tar 文件中提取到工作线程的移动有关线程。
另一方面，如何进行堆转储？

标签： java memory-leaks jaxb compression project-reactor

【解决方案1】：

经过几天的研究，事实证明根本没有内存泄漏。问题是由反应流的缓冲性质引起的。因此，在添加了一些跟踪之后，我发现在一个文件被读取和下沉之间可能有数百个正在飞行的对象。对于大多数情况，这是一个合理的性能优化，但在这个特殊的情况下，对象太重（100 个文件可能很容易达到 8GB）

我首先做的是使用 System.property reactor.bufferSize.small 将缓冲区大小减小到 16（可能的最小值）。

虽然这还不够，因为并行步骤。所以每个并行线程都向 Producer 询问 16 个文件 * 12 个线程（在我的 macbook 上）= 192 个飞行文件。为了避免这种情况，我们可以如下限制每个线程的获取率

.runOn(parallel(), 16 / DEFAULT_POOL_SIZE)

因此，我们确保此步骤一次在内存中的文件不超过 16 个。这样我终于可以在不损失任何处理能力的情况下将内存窥视保持在 4GB 以下。

【讨论】：