ExecutorServiceの終了に関する完了可能な将来の問題

コードにはいくつかの問題があります。completableFutureList後で追加される先物がある可能性がある場合のスナップショットの完了時にexecutorサービスのシャットダウンをスケジュールしていますが、さらに悪いことに// do something with pages、スナップショットさえまだ完了していない時点に達しています。

の宣言は示していませんがcompletableFutureList、異なるスレッドから変更したpagesand がスレッドセーフではないでsubmittedPages初期化されていることを考えると、HashSetリストについても気分がよくありません。しかし、とにかくリストは必要ありません。送信コードを変更して、後続のタスクで構成される保留中のタスクを表すフューチャーを返す必要があります。渡された関数はthenCompose、前提条件の段階が完了したときに評価されます。つまり、これにより、関数のチェーン時に不明な先物への依存が可能になります。

HashSetsをスレッドセーフ構造に置き換えるだけでは不十分であることに注意してください。他のスレッドがこれら2つの呼び出しの間に実行しないことが保証されていないため（「check-then-act」アンチパターンと呼ばcontainsれる）add、beforeのようなシーケンスを回避する必要がありaddます。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);
    }
}