时间戳

首页 > 解决方案 > 终止 ExecutorService 的可完成的未来问题

java - 终止 ExecutorService 的可完成的未来问题

问题描述

做一些基本的网络爬虫。

我想使用 Completable future 与多个线程并行运行抓取。每个作业都会检索需要抓取的 Page 对象，并返回带有已创建 url 列表的 Page 对象。

如果列表中的每个 url 尚未提交进行抓取，则它会启动新作业。完成所有并行作业后，我想继续进行逻辑。

如果我删除“allFutures.thenRun(() -> executorService.shutdown());”，此代码仅刮取第一页对象然后终止的问题 然后它收集所有页面/网址，但程序永远不会结束。

public class Demo
{
    private final Set<Page> pages = new HashSet<>();
    private final Set<Page> submittedPages = new HashSet<>();

    private final ExecutorService executorService;

    public Demo(final int numberOfThreads)
    {
        this.executorService = Executors.newFixedThreadPool(numberOfThreads);
    }

    public void start(String url) throws ExecutionException, InterruptedException
    {
        this.submitTask(new Page(url));
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[completableFutureList.size()]));
        allFutures.thenRun(() -> executorService.shutdown());

        // do something with pages
    }


    private void submitTask(final Page page)
    {
        if (!this.submittedPages.contains(page))
        {
            this.submittedPages.add(page);
            CompletableFuture<Void> cf = CompletableFuture.supplyAsync(() -> new Task(page).call(), this.executorService) //want to run this parallel in multiple threads
               .thenAccept(receivedPage -> {
                   this.savePage(receivedPage);
                   this.submitCollectedLinks(receivedPage);
               });
            completableFutureList.add(cf);

        }
    }

    private void submitCollectedLinks(final Page page){
        page.getLinks()
          .stream()
          .map(Page::new)
          .forEach(this::submitTask);
    }

    private void savePage(final Page page)
    {
        this.pages.add(page);
    }

}

标签： javacompletable-future

解决方案

您的代码有几个问题。您正在安排在完成快照后关闭执行程序服务，completableFutureList稍后可能会添加更多期货，但更糟糕的是，您甚至到达// do something with pages了快照尚未完成的地步。

您没有显示 的​​声明completableFutureList，但是鉴于您从不同线程修改的pagesand被初始化为，这不是线程安全的，我对列表也没有什么好感。但是无论如何您都不需要该列表。您应该更改提交代码以返回代表由后续任务组成的未决任务的期货。传递给的函数将在先决条件阶段完成后进行评估，换句话说，这允许在链接函数时依赖于未知的期货。submittedPagesHashSetthenCompose

请注意，用HashSet线程安全构造替换 s 是不够的。您必须避免像 call containsbefore这样的序列add，因为不能保证add在这两个调用之间没有其他线程会执行（称为“check-then-act”反模式）。你可以使用 just ，当元素已经存在时add它什么都不做并返回。false使用正确的线程安全Set实现，它提供了所需的原子性。

把这些东西放在一起，你会得到，例如

public class Demo {
    private final Set<Page> pages = ConcurrentHashMap.newKeySet();
    private final Set<Page> submittedPages = ConcurrentHashMap.newKeySet();

    private final ExecutorService executorService;

    public Demo(final int numberOfThreads) {
        this.executorService = Executors.newFixedThreadPool(numberOfThreads);
    }

    public void start(String url) {
        this.submitTask(new Page(url))
            // shutdown even in the exceptional case
            .whenComplete((_void, throwable) -> executorService.shutdown())
            .join(); // wait for completion before doing something with pages

        // do something with pages
    }

    private CompletableFuture<Void> submitTask(final Page page) {
        // use a single add to avoid check-then-act anti-pattern
        if(this.submittedPages.add(page)) {
            return CompletableFuture.supplyAsync(new Task(page)::call, executorService)
                // compose with recursively encountered tasks
               .thenCompose(receivedPage -> {
                   this.savePage(receivedPage);
                   return this.submitCollectedLinks(receivedPage);
               });
        }

        // do nothing when already submitted
        return CompletableFuture.completedFuture(null);
    }

    private CompletableFuture<Void> submitCollectedLinks(final Page page) {
        return CompletableFuture.allOf(page.getLinks()
          .stream().map(Page::new).map(this::submitTask)
          .toArray(CompletableFuture<?>[]::new));
    }

    private void savePage(final Page page) {
        this.pages.add(page);
    }
}

推荐阅读