ExecutorServiceは決して停止しません。他の実行中のタスクの中で新しいタスクを実行する場合

Question

おはよう。

私のウェブクローラープロジェクトでブロッカーの問題があります。 ロジックは簡単です。まずRunnableを1つ作成し、html文書をダウンロードしてすべてのリンクをスキャンし、すべての資金提供リンクに新しいRunnableオブジェクトを作成します。新しく作成されたそれぞれのRunnableは、順番に各リンクの新しいRunnableオブジェクトを作成し、実行します。

問題はExecutorServiceが決して停止しないということです。

CrawlerTest.java

public class CrawlerTest { 

    public static void main(String[] args) throws InterruptedException { 
     new CrawlerService().crawlInternetResource("https://jsoup.org/"); 
    } 
} 

CrawlerService.java

import java.io.IOException; 
import java.util.Collections; 
import java.util.Set; 
import java.util.concurrent.ConcurrentHashMap; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 
import java.util.concurrent.TimeUnit; 

import org.jsoup.Jsoup; 
import org.jsoup.nodes.Document; 
import org.jsoup.nodes.Element; 
import org.jsoup.select.Elements; 

public class CrawlerService { 

    private Set<String> uniqueUrls = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>(10000)); 
    private ExecutorService executorService = Executors.newFixedThreadPool(8); 
    private String baseDomainUrl; 

    public void crawlInternetResource(String baseDomainUrl) throws InterruptedException { 
     this.baseDomainUrl = baseDomainUrl; 
     System.out.println("Start"); 
     executorService.execute(new Crawler(baseDomainUrl)); //Run first thread and scan main domain page. This thread produce new threads. 
     executorService.awaitTermination(10, TimeUnit.MINUTES); 
     System.out.println("End"); 
    } 

    private class Crawler implements Runnable { // Inner class that encapsulates thread and scan for links 

     private String urlToCrawl; 

     public Crawler(String urlToCrawl) { 
      this.urlToCrawl = urlToCrawl; 
     } 

     public void run() { 
      try { 
       findAllLinks(); 
      } catch (InterruptedException e) { 
       e.printStackTrace(); 
      } 
     } 

     private void findAllLinks() throws InterruptedException { 
      /*Try to add new url in collection, if url is unique adds it to collection, 
      * scan document and start new thread for finded links*/ 
      if (uniqueUrls.add(urlToCrawl)) { 
       System.out.println(urlToCrawl); 

       Document htmlDocument = loadHtmlDocument(urlToCrawl); 
       Elements findedLinks = htmlDocument.select("a[href]"); 

       for (Element link : findedLinks) { 
        String absLink = link.attr("abs:href"); 
        if (absLink.contains(baseDomainUrl) && !absLink.contains("#")) { //Check that we are don't go out of domain 
         executorService.execute(new Crawler(absLink)); //Start new thread for each funded link 
        } 
       } 
      } 
     } 

     private Document loadHtmlDocument(String internetResourceUrl) { 
      Document document = null; 
      try { 
       document = Jsoup.connect(internetResourceUrl).ignoreHttpErrors(true).ignoreContentType(true) 
         .userAgent("Mozilla/5.0 (Windows NT 6.1; WOW64; rv:48.0) Gecko/20100101 Firefox/48.0") 
         .timeout(10000).get(); 
      } catch (IOException e) { 
       System.out.println("Page load error"); 
       e.printStackTrace(); 
      } 
      return document; 
     } 
    } 
} 

このアプリは、すべてのユニークなリンクについてjsoup.orgをスキャンするのに約20秒を必要としています。しかし、それはちょうど10分待つexecutorService.awaitTermination(10, TimeUnit.MINUTES); そして私は死んだメインスレッドとまだ実行中のエグゼキュータを参照してください。正しくExecutorService仕事を強制する方法

Threads

？

私はexecutorService.executeをメインスレッドではなく別のタスクの内部で呼び出すことが問題だと思います。

Answer 1

私は以前から、あなたのコメントを参照してください。私はリソースから収集しますどのように多くのユニークなリンクが事前にわからないので、私はされたCountDownLatchを使用することはできません

を。

まず、vsminkovはなぜawaitTermniationが座って10分間待つのかという答えを見つけてくれます。私は別の解決法を提供します。

CountDownLatchを使用する代わりに、Phaserを使用してください。新しいタスクごとに、登録して完了を待つことができます。

execute.submitが呼び出されるたびに1つの位相器とregisterを作成し、Runnableが完了するたびにarriveを作成します。正しくExecutorServiceの作業を強制する方法

public void crawlInternetResource(String baseDomainUrl) { 
    this.baseDomainUrl = baseDomainUrl; 

    Phaser phaser = new Phaser(); 
    executorService.execute(new Crawler(phaser, baseDomainUrl)); 
    int phase = phaser.getPhase(); 
    phase.awaitAdvance(phase); 
} 

private class Crawler implements Runnable { 

    private final Phaser phaser; 
    private String urlToCrawl; 

    public Crawler(Phaser phaser, String urlToCrawl) { 
     this.urlToCrawl = urlToCrawl; 
     this.phaser = phaser; 
     phaser.register(); // register new task 
    } 

    public void run(){ 
     ... 
     phaser.arrive(); //may want to surround this in try/finally 
    } 

Answer 2

あなたは誤ってawaitTerminationです。すべてのタスクがシャットダウン要求後に実行が完了した、またはタイムアウトが発生した場合、または現在のスレッドが中断され、いずれか早い方まで

ブロックを：javadocツールによると、あなたはshutdown最初に呼び出す必要があります。私はあなたが安全にshutdownを行うことができますので、残されたタスクがない場合、正確な瞬間を決定するためにCountDownLatchを使用（またはthis oneのようなその支援増分をラッチ）することをお勧めしたいあなたの目標を達成するために

。

Answer 3

シャットダウンを呼び出していません。

これは機能する可能性があります - CrawlerServiceのAtomicLong変数。新しいサブタスクがすべて実行者サービスに提出される前にインクリメントされます。

このカウンタをデクリメントするためにあなたのrun（）メソッドを変更している場合0、シャットダウン「ついに」でキュータサービス

public void run() { 
    try { 
     findAllLinks(); 
    } catch (InterruptedException e) { 
     e.printStackTrace(); 
    } finally { 
     //decrements counter 
     //If 0, shutdown executor from here or just notify CrawlerService who would be doing wait(). 
    } 
} 

、カウンタを削減し、カウンタがゼロ、シャットダウンエグゼキュータまたは単に通知しているときCrawlerService。 0は、これが最後のものであり、他のキューは実行されておらず、待ち行列には保留されていないことを意味します。タスクは新しいサブタスクを提出しません。

Answer 4

？

私はexecutorService.executeをメインスレッドではなく別のタスクの内部で呼び出すことが問題だと思います。

いいえ問題はExecutorServiceではありません。誤った方法でAPIを使用しているため、正しい結果が得られません。

正しい結果を得るには、特定の順序で3つのAPIを使用する必要があります。 ExecutorServiceのOracleのドキュメントページから

1. shutdown 
2. awaitTermination 
3. shutdownNow 

推奨される方法：

void shutdownAndAwaitTermination(ExecutorService pool) { 
    pool.shutdown(); // Disable new tasks from being submitted 
    try { 
    // Wait a while for existing tasks to terminate 
    if (!pool.awaitTermination(60, TimeUnit.SECONDS)) { 
     pool.shutdownNow(); // Cancel currently executing tasks 
     // Wait a while for tasks to respond to being cancelled 
     if (!pool.awaitTermination(60, TimeUnit.SECONDS)) 
      System.err.println("Pool did not terminate"); 
    } 
    } catch (InterruptedException ie) { 
    // (Re-)Cancel if current thread also interrupted 
    pool.shutdownNow(); 
    // Preserve interrupt status 
    Thread.currentThread().interrupt(); 
    } 

shutdown():は以前に提出したタスクが実行される通常のシャットダウンを開始しますが、新しいタスクは受け入れられません。

shutdownNow():アクティブに実行中のタスクをすべて停止しようとし、待機中のタスクの処理を停止し、実行を待っていたタスクのリストを返します。

awaitTermination():シャットダウン要求後にすべてのタスクが実行を完了するか、タイムアウトが発生するか、または現在のスレッドが中断されるまでのいずれか早いタイミングで実行されるまでブロックします。

別のノートで

：

wait until all threads finish their work in java

私はinvokeAll()またはForkJoinPool()を使用して好む、あなたの使用のために最も適している：あなたはすべてのタスクが完了するのを、この関連のSEの質問を参照して、待ちたい場合場合。