Apache Fury 反序列化时处理慢速输入流的优化方案

Apache Fury 是一个高性能的序列化框架，但在处理慢速输入流（如 GZIP 压缩流）时可能会遇到问题。本文将深入分析这个问题及其解决方案。

问题背景

当使用 Fury 框架从 GZIP 压缩的输入流中反序列化多个对象时，可能会遇到 IllegalArgumentException 异常。这是因为 Fury 的 readToBufferFromStream 方法假设输入流能够一次性读取完整的序列化数据，而实际上 GZIP 等压缩流可能会分批返回数据。

问题重现

考虑以下典型使用场景：

try (InputStream input = new GZIPInputStream(new FileInputStream("test.dat.gz"))) {
  while (input.available() > 0) {
    MyClass obj = (MyClass) fury.deserialize(input);
  }
}

当输入流无法一次性提供完整数据时，Fury 会抛出异常，因为现有的实现要求必须完整读取4字节的长度信息和后续的对象数据。

技术分析

问题的根源在于 readToBufferFromStream 方法的实现过于理想化，没有考虑输入流可能分批返回数据的情况。原始实现直接调用 inputStream.read() 并期望一次性读取所有数据，这在网络传输或压缩流等场景下往往不成立。

解决方案

通过引入一个辅助方法 readBytes，可以实现对输入流的分批读取：

private static void readToBufferFromStream(InputStream inputStream, MemoryBuffer buffer)
    throws IOException {
  buffer.readerIndex(0);
  int read = readBytes(inputStream, buffer.getHeapMemory(), 0, 4);
  Preconditions.checkArgument(read == 4);
  int size = buffer.readInt();
  buffer.ensure(4 + size);
  read = readBytes(inputStream, buffer.getHeapMemory(), 4, size);
  Preconditions.checkArgument(read == size);
}

private static int readBytes(InputStream inputStream, byte[] buffer, 
    int offset, int size) throws IOException {
  int read = 0;
  int count = 0;
  while (read < size) {
    if ((count = inputStream.read(buffer, offset + read, size - read)) == -1) {
      break;
    }
    read += count;
  }
  return (read == 0 && count == -1) ? -1 : read;
}

这个改进方案具有以下特点：

1. 使用循环读取确保获取足够的数据量
2. 正确处理流结束的情况（返回-1）
3. 保持原有的参数校验逻辑
4. 兼容各种类型的输入流，包括慢速流

实际应用价值

这个改进使得 Fury 框架能够更好地处理以下场景：

• 从压缩流（GZIP、ZIP等）中反序列化数据
• 网络传输场景下的分批数据传输
• 大对象的分块读取
• 任何可能产生数据延迟的输入源

总结

通过对 Fury 反序列化过程的这一优化，显著提高了框架在处理慢速输入流时的健壮性和兼容性。这种改进对于需要处理压缩数据或网络传输数据的应用场景尤为重要，确保了 Fury 在各种复杂环境下都能可靠工作。