ExecutorService가 중지되지 않습니다. 실행중인 다른 작업에서 새 작업을 실행할 때

Question

안녕하세요.

웹 크롤러 프로젝트에 방해물 문제가 있습니다. 논리가 간단합니다. 먼저 Runnable 하나를 만들고 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); 그리고 난 죽은 메인 쓰레드를보고 여전히 실행 중이다.

Threads

어떻게 제대로 ExecutorService 작업을 강제로?

문제는 executorService.execute가 메인 스레드 대신 다른 태스크 내부에서 호출된다는 것입니다.

Answer 1

나는 이전부터 의견을 참조하십시오 내가 리소스에서 수집합니다 사전에 얼마나 많은 고유 링크 모르기 때문에

가

내가 CountDownLatch를 사용할 수 없습니다.

우선 vsminkov는 awaitTermniation이 앉아서 10 분을 기다리는 이유에 대한 답을 확인했습니다. 대체 솔루션을 제공 할 것입니다.

CountDownLatch을 사용하는 대신 Phaser을 사용하십시오. 새로운 작업마다 등록하고 완료를 기다릴 수 있습니다.

하나의 페이저를 만들고 register마다 execute.submit가 호출되고 arrive마다 Runnable 완료.

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. 모든 작업이 종료 요청 후 실행을 완료 할 때까지

블록, 또는 타임 아웃이 발생하거나, 현재의 thread는 둘 중 먼저 발생, 중단 : javadoc에 따르면 당신은 shutdown 먼저 호출해야합니다.

내가 CountDownLatch를 사용하는 것이 좋을 것 당신의 목표를 달성 (또는 this one 같은 지원하는 단위 래치) 안전하게 shutdown을 할 수 있도록 남은 작업이 없을 때 정확한 순간을 결정합니다.

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(). 
    } 
} 

, 카운터를 감소시키고 카운터가 0 종료 집행하거나 알릴 때 CrawlerService. 0은 이것이 마지막 것임을 의미하고 다른 하나는 실행 중이 지 않으며 대기열에 대기 중이 아닙니다. 어떤 작업도 새로운 하위 작업을 제출하지 않습니다.

Answer 4

어떻게 제대로 ExecutorService를 작업을 강제로?

문제는 executorService.execute가 메인 스레드 대신 다른 태스크 내부에서 호출된다는 것입니다.

아니요. 문제는 ExecutorService가 아닙니다. 잘못된 방식으로 API를 사용하고 있으므로 올바른 결과를 얻지 못합니다.

올바른 결과를 얻으려면 특정 순서로 3 개의 API를 사용해야합니다. ExecutorService의 오라클 문서 페이지에서

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 질문을 참조 기다려야합니다 케이스.