java - 使用 Java 8 流映射函数来执行长时间运行的任务

Question

我有一个方法，它将(远程文件的)URL 列表作为应该下载的参数。该方法返回一个其他类型的List(称为Attachment)，它实际上包含java File类型的属性。 对于本例，我使用 Java Stream API 迭代 URL 并在“map”函数中启动下载，该函数实际上返回附件实例。

现在我的问题是:我是否滥用 Java Stream API 来做一些不该做的事情？喜欢将长时间运行任务放入其中吗？我应该只对输入数据进行小操作吗？

我现在看到的唯一缺点是测试有点困难。

private List<Attachment> download(List<URL> attachments) {
        return attachments.stream().map(attachmentUrl -> {
            try {
                Attachment attachment = new Attachment();
                File attachmentFile = new File(getFilename(attachment.getAttachmentId(), attachmentUrl));
                FileUtils.copyURLToFile(
                        attachmentUrl,
                        attachmentFile,
                        CONNECT_TIMEOUT,
                        READ_TIMEOUT);
                attachment.setAttachmentFile(attachmentFile);
                return attachment;
            } catch (IOException e) {
                e.printStackTrace();
                LOGGER.error(e.getLocalizedMessage());
            }
            return null;
        }).filter(Objects::nonNull).collect(Collectors.toList());
    }

Answer 1

我认为考虑 map 和其他功能结构(例如 filter、reduce 等)可能会有所帮助。作为函数，而是作为语法。 stream().map() 是执行与 for 循环功能等效的语法。问“我是否因为使用该语法执行的内容而滥用了该语法？”那么就没那么有意义了:for 循环不关心它们在每次迭代中运行的任务需要多长时间，map 也不关心。它与它所应用的操作无关，因此唯一的问题是您是否正确使用了语法，即循环遍历集合，映射从某物到某物。

在这种情况下，其中 map 是语法，您想要的操作完全没问题。但是，您的实现可以稍微清理一下。

attachmentUrl -> {
    try {
        Attachment attachment = new Attachment();
        File attachmentFile = new File(getFilename(attachment.getAttachmentId(), attachmentUrl));
        FileUtils.copyURLToFile(
                attachmentUrl,
                attachmentFile,
                CONNECT_TIMEOUT,
                READ_TIMEOUT);
        attachment.setAttachmentFile(attachmentFile);
        return attachment;
    } catch (IOException e) {
        e.printStackTrace();
        LOGGER.error(e.getLocalizedMessage());
    }
    return null;
}

对于内联 map lambda 来说有点大。一般来说，我倾向于对任何需要大括号(即占用多行)的 map lambda 持怀疑态度，尽管并不总是反对。我建议将此 lambda 重构为命名函数，并且可能是一对，它们要么是嵌套的 (map(this::A)，其中 A 然后调用 B) 或由流操作连续使用 map(this::A).map(this::B)。

<小时/>

[编辑:]关于并行化您的流:请记住，作为此方法的一部分，您所做的不仅仅是CPU处理 - 您似乎正在执行网络IO和 文件 IO。如果并行执行，您不仅会并行化 CPU 利用率，还会并行化网络和磁盘使用率。如果网络或磁盘是主导因素，而不是CPU，那么并行化可能不会给你带来什么好处，而且可能会让事情变得更糟。一般来说，更多的线程!=更快的网络或磁盘读/写。您可能会发现this question on parallel IO有用。