parallelStreamsの実行中にどのように待機するのですか？

Question

だから、今対処しましょう。私はhtmlからリンクを得ることができるParserクラスを持っています。 Jsoupを使用してリンクを取得した後、このリンクをListに追加します。私はStreamAPIを使ってそれを進めています。 次のコードを見てください：

public static List<Link> getLinksFromURL(String URL) { 
    List<Link> links = new ArrayList<>(); 
    Elements elements = selectElements(URL, "a[href]"); 
    elements.parallelStream().forEach((Parser) -> addLink(links, elements.attr("href"))); 
    return links; 
} 

これは機能します。しかし、私は何を扱っていますか？このメソッドをテストしたところ、ストリームの実行はしばらくの間終了することがあります。私はテストクラスで間違ったデータを得ることができます。例えば。リストのサイズをチェックします。デバッグ中は、すべて正常です。しかし、実行時には、ストリームがまだ実行されていないため、すべてのリンクではなく、いくつかの要素を取得します。 シンプルなテスト。しかし、あまりいつかサイズ（）== 3か：

@Test 
public void getLinksFromURLTest() { 
    Assert.assertTrue(Parser.getLinksFromURL("testLinks").size()==4); 
} 

そして、私の質問は、このストリームが実行している間、私は待つことができるかのですか？私はすべてのリンクを取得する必要があります:)

小規模な注意：私はテストのために立ち上げるスパークサーバーを表すローカルHTMLからhtmlを取得します。

P .:私が理解できない場合は、教えてください。私に説明を追加してください。

私を助けてくれたら、とても感謝しています。 幸運な皆さん！ :)

更新：doc 1として addLink方法

private static void addLink(List<Link> links, String URL) { 
    if (!URL.isEmpty() && isLink(URL) && !hasSameLink(links, URL)){ 
     links.add(new Link(URL)); 
    } 
} 

Answer 1

あなたはforEach()コールで間違いをしています。各要素について、Element.attr()の代わりにelements変数のElements.attr()を呼び出しています。

forEach(element -> addLink(links, element.attr("href"))) 

これに関係なく、コードはスレッドセーフではありません。 ArrayListに複数のスレッドを書き込むことはできません（これは並列化の利点を失う可能性があります）。これはまた、あなたのテストで一貫性のない結果が出ている理由です。スレッドセーフなコレクションを使用するか、連続した繰り返しを持つスティックのみを使用する必要があります。

また、あなたの代わりにコレクタをストリームパイプラインにあなたのロジックのすべてを変換して使用することができる場合があります

return elements.parallelStream() 
     .filter(url -> !url.isEmpty()) 
     .filter(url -> isLink(url)) 
     .distinct() 
     .map(Link::new) 
     .collect(Collectors.toList()); 

Answer 2

、それはコードを同期するのはあなた次第です。要素数でCountdownLatchを初期化してから、メインスレッドで待機させることができます。