java - 将列表值顺序传递给单值消费者的最佳方式？

Question

我正在研究 Java8 的流和 CompletableFuture秒。我预先存在的代码有一个类，它接受一个 URL 并下载它:

public class FileDownloader implements Runnable {
    private URL target;
    public FileDownloader(String target) {
        this.target = new URL(target);
    }
    public void run() { /* do it */ }
}

现在，此类从另一个发出 List<String> 的部分获取信息(单个主机上的多个目标)。

我已将周围的代码切换为 CompletableFuture :

public class Downloader {
    public static void main(String[] args) {
        List<String> hosts = fetchTargetHosts();
        for (String host : hosts) {
            HostDownloader worker = new HostDownloader(host);
            CompletableFuture<List<String>> future = 
                CompletableFuture.supplyAsync(worker);
            future.thenAcceptAsync((files) -> {
                for (String target : files) {
                    new FileDownloader(target).run();
                }
            });
        }
    }

    public static class HostDownloader implements Supplier<List<String>> {
        /* not shown */ 
    }
    /* My implementation should either be Runnable or Consumer.
       Please suggest based on a idiomatic approach to the main loop.
     */
    public static class FileDownloader implements Runnable, Consumer<String> { 
        private String target;
        public FileDownloader(String target) {
            this.target = target;
        }

        @Override
        public void run() { accept(this.target); }

        @Override
        public void accept(String target) {
            try (Writer output = new FileWriter("/tmp/blubb")) {
                output.write(new URL(target).getContent().toString());
            } catch (IOException e) { /* just for demo */ }
        }
    }
}

现在，这感觉不自然。我正在制作 String 的流我和我的 FileDownloader一次消耗其中一个。有没有现成的启用我的单值Consumer与List一起工作还是我坚持使用 for在这里循环？

我知道将循环移动到 accept 中是微不足道的然后做一个Consumer<List<String>> ，这不是重点。

Answer 1

将两个直接相关的步骤分解为两个异步步骤是没有意义的。他们仍然相互依赖，如果分离有任何影响，那也不会是积极的。

你可以简单地使用

List<String> hosts = fetchTargetHosts();
FileDownloader fileDownloader = new FileDownloader();
for(String host: hosts)
    CompletableFuture.runAsync(()->
        new HostDownloader(host).get().forEach(fileDownloader));

或者，假设 FileDownloader 没有关于下载的可变状态:

for(String host: hosts)
    CompletableFuture.runAsync(()->
        new HostDownloader(host).get().parallelStream().forEach(fileDownloader));

这仍然具有与使用 supplyAsync 加上 thenAcceptAsync 的原始方法相同的并发级别，只是因为这两个相关步骤无论如何都不能同时运行，所以简单解决方案是将这两个步骤放在一个将异步执行的简洁操作中。

---

但是，此时值得注意的是，不建议将 CompletableFuture 全部用于此操作。作为it’s documentation状态:

• All async methods without an explicit Executor argument are performed using the ForkJoinPool.commonPool()

common pool 的问题是它预先配置的并发级别取决于 CPU 内核的数量，如果在 I/O 操作期间线程被阻塞，则不会调整。换句话说，它不适用于 I/O 操作。

与 Stream 不同，CompletableFuture 允许您为 async 操作指定一个 Executor，因此您可以配置你自己的 Executor 适合 I/O 操作，另一方面，当你处理一个 Executor 时，就不需要 CompletableFuture 根本没有，至少不是为了这样一个简单的任务:

List<String> hosts = fetchTargetHosts();

int concurrentHosts = 10;
int concurrentConnections = 100;
ExecutorService hostEs=Executors.newWorkStealingPool(concurrentHosts);
ExecutorService connEs=Executors.newWorkStealingPool(concurrentConnections);

FileDownloader fileDownloader = new FileDownloader();
for(String host: hosts) hostEs.execute(()-> {
    for(String target: new HostDownloader(host).get())
        connEs.execute(()->fileDownloader.accept(target));
});

在这个地方，您可以考虑将 FileDownloader.accept 的代码内联到 lambda 表达式中，或者将其恢复为 Runnable，以便您可以更改connEs.execute(new FileDownloader(target)) 的内部循环语句。