我必须迭代 130 多个数据传输对象，每次都会生成一个 json 上传到 aws S3。

在没有改进的情况下，完成整个过程大约需要 90 秒。我尝试使用 lamba 而不是使用 lamba，两者的结果相同。

for(AbstractDTO dto: dtos) {
    try {
        processDTO(dealerCode, yearPeriod, monthPeriod, dto);
    } catch (FileAlreadyExistsInS3Exception e) {
        failedToUploadDTOs.add(e.getLocalizedMessage() + ": " + dto.fileName() + ".json");
    }
}

dtos.stream().forEach(dto -> {
    try {
        processDTO(dealerCode, yearPeriod, monthPeriod, dto);
    } catch (FileAlreadyExistsInS3Exception e) {
        failedToUploadDTOs.add(e.getLocalizedMessage() + ": " + dto.fileName() + ".json");
    }
});

经过一些调查，我得出结论，processDTO 方法每项运行大约需要 0.650 毫秒。

我的第一次尝试是使用并行流，结果还不错，用了大约15秒完成整个过程:

dtos.parallelStream().forEach(dto -> {
    try {
        processDTO(dealerCode, yearPeriod, monthPeriod, dto);
    } catch (FileAlreadyExistsInS3Exception e) {
        failedToUploadDTOs.add(e.getLocalizedMessage() + ": " + dto.fileName() + ".json");
    }
});

但我仍然需要减少那个时间。 我研究了如何改进并行流，并发现了 ForkJoinPool 技巧:

ForkJoinPool forkJoinPool = new ForkJoinPool(PARALLELISM_NUMBER);
forkJoinPool.submit(() ->
dtos.parallelStream().forEach(dto -> {
    try {
        processDTO(dealerCode, yearPeriod, monthPeriod, dto);
    } catch (FileAlreadyExistsInS3Exception e) {
        failedToUploadDTOs.add(e.getLocalizedMessage() + ": " + dto.fileName() + ".json");
    }
})).get();
forkJoinPool.shutdown();

不幸的是，结果让我有点困惑。

• 当 PARALLELISM_NUMBER 为 8 时，完成整个过程大约需要 13 秒。改善不大。
• 当 PARALLELISM_NUMBER 为 16 时，完成整个过程大约需要 8 秒。
• 当 PARALLELISM_NUMBER 为 32 时，完成整个过程大约需要 5 秒。

所有测试都是使用 postman 请求完成的，调用 Controller 方法最终将迭代 130 个项目

我对 5 秒感到满意，使用 32 作为 PARALLELISM_NUMBER，但我担心后果。

• 保留 32 可以吗？
• 理想的 PARALLELISM_NUMBER 是多少？
• 在决定它的值(value)时我必须牢记什么？

我在 Mac 2.2GHZ I7 上运行

sysctl hw.physicalcpu hw.logicalcp
hw.physicalcpu: 4
hw.logicalcpu: 8

processDTO 的作用如下:

private void processDTO(int dealerCode, int yearPeriod, int monthPeriod, AbstractDTO dto) throws FileAlreadyExistsInS3Exception {
    String flatJson = JsonFlattener.flatten(new JSONObject(dto).toString());
    String jsonFileName = dto.fileName() + JSON_TYPE;;
    String jsonFilePath = buildFilePathNew(dto.endpoint(), dealerCode, yearPeriod, monthPeriod, AWS_S3_JSON_ROOT_FOLDER);
    uploadFileToS3(jsonFilePath + jsonFileName, flatJson);
}

public void uploadFileToS3(String fileName, String fileContent) throws FileAlreadyExistsInS3Exception {
    if (s3client.doesObjectExist(bucketName, fileName)) {
        throw new FileAlreadyExistsInS3Exception(ErrorMessages.FILE_ALREADY_EXISTS_IN_S3.getMessage());
    }
    s3client.putObject(bucketName, fileName, fileContent);
}

最佳答案

parallelism 参数决定了ForkJoinPool 使用多少线程。这就是为什么默认情况下 parallelism 值是可用的 CPU 核心数:

Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors())

在您的情况下，瓶颈应该是检查文件是否存在并将其上传到 S3。这里的时间将取决于至少几个因素:CPU、网卡和驱动程序、操作系统、其他。在您的情况下，S3 网络操作时间似乎不受 CPU 限制，因为您通过创建更多模拟工作线程观察到改进，也许网络请求已由操作系统排队。

parallelism 的正确值因工作负载类型而异。由于上下文切换的负面影响，默认 parallelism 等于 CPU 核心的 CPU 绑定(bind)工作流更好。假设工作负载不会阻塞 CPU，像您这样的非 CPU 绑定(bind)工作负载可以通过更多工作线程来加速，例如通过 busy waiting .

ForkJoinPool 中的parallelism 没有单一的理想值。

关于java - 如何提高迭代 130 多个项目并将它们上传到 aws s3 的性能，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/55500189/