Многопоточность Java, выполняющая больше, чем ограничения цикла

Question

Я создаю веб-скребок, чтобы вытащить как ссылки, так и электронные письма из Интернета. Ссылки будут использоваться для поиска новых мест для поиска электронных писем, и электронные письма затем будут сохранены в наборе. Каждая ссылка передается в фиксированный пул потоков в своем потоке, чтобы искать больше писем. Я начал с малого только на поиск 10 писем, но по какой-то причине мой код возвращает около 13 писем.

while (emailSet.size() <= EMAIL_MAX_COUNT) { 

     link = linksToVisit.poll(); 

     linksToVisit.remove(link); 
     linksVisited.add(link); 
     pool.execute(new Scraper(link)); 
    } 

    pool.shutdownNow(); 

    emailSet.stream().forEach((s) -> { 
     System.out.println(s); 
    }); 
    System.out.println(emailSet.size()); 

Хотя я понимаю, что можно создавать дополнительные потоки, которые будут по-прежнему завершенными после я получаю 10 сообщений электронной почты не должны pool.shutdownNow() закончить эти темы?

Вот мой код темы, если это помогает.

class Scraper implements Runnable { 

    private String link; 

    Scraper(String s) { 
     link = s; 
    } 

    @Override 
    public void run() { 
     try { 
      Document doc = (Document) Jsoup.connect(link).get(); 

      Elements links = doc.select("a[href]"); 
      for (Element href : links) { 
       String newLink = href.attr("abs:href"); 
       if (!linksVisited.contains(newLink) && !linksToVisit.contains(newLink)) { 
        linksToVisit.add(newLink); 
       } 
      } 
      Pattern p = Pattern.compile(
        "[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\\.[a-zA-Z0-9-.]+"); 
      Matcher matcher = p.matcher(doc.text()); 
      while (matcher.find()) { 
       emailSet.add(matcher.group()); 
      } 
     } catch (Exception e) { 
      //Catch on of the many exceptions Jsoup.connect might throw 
      // and just let the thread expire. 
     } 
    } 
} 

Edit 1:

я включённое это мой первый раз, но я использую Потокобезопасную набор и очередь.

Set<String> emailSet = Collections.synchronizedSet(new HashSet()); 
BlockingQueue<String> linksToVisit = new ArrayBlockingQueue(10000); 
Set<String> linksVisited = Collections.synchronizedSet(new HashSet()); 
final int EMAIL_MAX_COUNT = 10; 
ExecutorService pool = newFixedThreadPool(25); 

Edit 2

Фигурные, что я должен обновить свой вопрос с ответом, так вот где моя проблема была.

while (emailSet.size() <= EMAIL_MAX_COUNT) { 

    link = linksToVisit.poll(); 

    linksToVisit.remove(link); 
    linksVisited.add(link); 
    pool.execute(new Scraper(link)); 
} 

Список будет начинаться только с одной ссылки. После удаления первой ссылки у меня был пустой список, который продолжал создавать новые потоки без ссылки для поиска. До того, как список мог быть заполнен, я уже создал сотни потоков, которые ничего не делали, но замедляли работу системы, пока она, наконец, не потерпит крах.

Вот исправление кода, чтобы гарантировать, что потоки не будут созданы, если не было ссылки для поиска.

while (emailSet.size() <= EMAIL_MAX_COUNT) { 

     if (linksToVisit.size() > 0) { 
      link = linksToVisit.poll(); 

      linksToVisit.remove(link); 
      linksVisited.add(link); 
      pool.execute(new Scraper(link)); 
      //System.out.println("Emails " + emailSet.size()); 
     } else { 
      try { 
       Thread.sleep(100); 
      } catch (InterruptedException ex) { 
       Logger.getLogger(Crawler.class.getName()) 
         .log(Level.SEVERE, null, ex); 
      } 
     } 
    } 

Answer 1

Вы начинаете scaper в асинхронном способе внутри цикла, который проверяет размер emailSet, но в течение одного цикла цикла скребок может найти более чем один адрес электронной почты, или вы можете начать более чем один скребок и после того, как вы начнете это добавляет алла электронной почты ссылки на страницу Рассмотрим следующий регламент

T1 loop start ->T2 loop schedule Scaper ->T3 loop check emailSet ->T4 Scraper finds 13 email -> T5 loop check emailSet 

или следующие один

T1 loop start ->T2 loop schedule Scaper "1" ->T3 loop check emailSet ->T4 loop schedule Scaper "2" T5 -> Scraper "1" finds 6 emails -> T6 loop check emailSet -> Scraper "1" finds 7 emails 

и скоро.

Если вы хотите остановиться, когда вы найдете 10 писем вы должны изменить следующие один

while (matcher.find()) { 
    emailSet.add(matcher.group()); 
} 

в

while (matcher.find()) { 
    if (emailSet.size() <= EMAIL_MAX_COUNT) { 
     emailSet.add(matcher.group()); 
    } 
} 

и даже это не полностью гарантирует, что вы можете остановиться на EMAIL_MAX_COUNT, потому что с несколькими потоками (например, 3), можно проверить размер и получить 9, а затем все они вставить электронное письмо.

Вы должны синхронизировать операцию чтения и записи в одном блоке (с synchronized(emailSet) или с помощью блокировки), если вы хотите обеспечить точный размер электронной почты; что-то вроде

while (matcher.find()) { 
    synchronized(emailSet) { 
     if (emailSet.size() <= EMAIL_MAX_COUNT) { 
      emailSet.add(matcher.group()); 
     } 
    } 
}