Java Streams：如何将Stream<InputStream>转为固定大小列表流优化PDF合并

核心逻辑是将有序的InputStream流按固定大小拆分批次，用批量合并方法处理每一批，再合并各批次的输出，既减少合并次数提升性能，也避免一次性加载过多输入导致的执行失败。

前置要求 #

先确保pdfService提供批量合并方法：InputStream mergePdf(List<InputStream> inputs)，这是分批方案性能收益的核心，比两两合并效率高很多。如果原本只有两两合并的方法，优先重载一个支持列表输入的版本。

代码实现 #

简化版本（依赖Guava，代码最简洁） #

如果你的项目已经引入Guava，直接用Lists.partition实现分批：

import com.google.common.collect.Lists;

// 按压测结果调整分批大小，建议初始值设为10~20，再根据实际性能微调
int BATCH_SIZE = 15;

// 先按顺序拿到所有InputStream（仅存引用，内存占用极低）
List<InputStream> allSortedStreams = dynamoDBItems.parallelStream()
        .map(dynamoDBItem -> new ChonkRequest(
                dynamoDBItem.getString("s3_key"),
                dynamoDBItem.getString("s3_bucket"),
                dynamoDBItem.getInt("SK")
        ))
        .sorted(Comparator.comparingInt(ChonkRequest::getIndex))
        .map(storageService::getChonk)
        .collect(Collectors.toList());

// 分批合并
InputStream pdfReportStream = Lists.partition(allSortedStreams, BATCH_SIZE)
        .stream()
        .reduce(null, (mergedStream, currentBatch) -> {
            try {
                List<InputStream> mergeInputs = new ArrayList<>();
                // 已合并的流非空则加入合并队列
                if (mergedStream != null) {
                    mergeInputs.add(mergedStream);
                }
                mergeInputs.addAll(currentBatch);
                // 执行批量合并
                InputStream newMergedStream = pdfService.mergePdf(mergeInputs);
                // 手动关闭已处理的流，避免资源泄漏
                if (mergedStream != null) {
                    mergedStream.close();
                }
                currentBatch.forEach(InputStream::close);
                return newMergedStream;
            } catch (IOException e) {
                throw new RuntimeException("PDF分批合并失败", e);
            }
        });

无依赖版本（自定义分批工具） #

如果不想引入Guava，自己实现分批工具方法即可，兼容Java 8及以上版本：

/**
 * 有序流按固定大小分批，保证元素顺序不变
 */
public static <T> Stream<List<T>> batchStream(Stream<T> source, int batchSize) {
    if (batchSize < 1) {
        throw new IllegalArgumentException("分批大小必须大于0");
    }
    Spliterator<T> sourceSpliterator = source.spliterator();
    if (!sourceSpliterator.hasCharacteristics(Spliterator.ORDERED)) {
        throw new IllegalArgumentException("仅支持顺序流分批，保证顺序正确");
    }
    Iterator<T> sourceIterator = Spliterators.iterator(sourceSpliterator);
    long estimatedBatchCount = sourceSpliterator.estimateSize() == Long.MAX_VALUE
            ? Long.MAX_VALUE
            : (sourceSpliterator.estimateSize() + batchSize - 1) / batchSize;

    Spliterator<List<T>> batchSpliterator = Spliterators.spliterator(
            new Iterator<>() {
                @Override
                public boolean hasNext() {
                    return sourceIterator.hasNext();
                }

                @Override
                public List<T> next() {
                    List<T> batch = new ArrayList<>(batchSize);
                    for (int i = 0; i < batchSize && sourceIterator.hasNext(); i++) {
                        batch.add(sourceIterator.next());
                    }
                    return batch;
                }
            },
            estimatedBatchCount,
            Spliterator.ORDERED | Spliterator.IMMUTABLE | Spliterator.NONNULL
    );
    return StreamSupport.stream(batchSpliterator, source.isParallel())
            .onClose(source::close);
}

业务代码调整为：

int BATCH_SIZE = 15;

InputStream pdfReportStream = dynamoDBItems.parallelStream()
        .map(dynamoDBItem -> new ChonkRequest(
                dynamoDBItem.getString("s3_key"),
                dynamoDBItem.getString("s3_bucket"),
                dynamoDBItem.getInt("SK")
        ))
        .sorted(Comparator.comparingInt(ChonkRequest::getIndex))
        .map(storageService::getChonk)
        .sequential() // 转串行流保证分批顺序正确
        .collect(Collectors.collectingAndThen(Collectors.toList(), list -> batchStream(list.stream(), BATCH_SIZE)))
        .reduce(null, (mergedStream, currentBatch) -> {
            // 合并逻辑和Guava版本完全一致
            try {
                List<InputStream> mergeInputs = new ArrayList<>();
                if (mergedStream != null) {
                    mergeInputs.add(mergedStream);
                }
                mergeInputs.addAll(currentBatch);
                InputStream newMergedStream = pdfService.mergePdf(mergeInputs);
                if (mergedStream != null) {
                    mergedStream.close();
                }
                currentBatch.forEach(InputStream::close);
                return newMergedStream;
            } catch (IOException e) {
                throw new RuntimeException("PDF分批合并失败", e);
            }
        });

注意事项 #

• 分批大小调优：分批大小直接决定性能，建议先在测试环境压测5、10、20、30几个档位，找到内存占用和执行速度的平衡点。
• 资源关闭：示例中已加入流关闭逻辑，如果你的mergePdf方法内部已经会自动关闭所有输入流，可以根据实际情况调整关闭逻辑，避免重复关闭报错。
• 并行优化：如果PDF合并是CPU密集型操作，分批完成后可以将流转为并行流执行reduce，只要保证分批阶段是顺序的，并行合并的结果不会出错。

内容的提问来源于stack exchange，提问作者firestreak