Почему выполняемая задача не может быть прервана с помощью флага AtomicBoolean?

Question

У меня есть следующая задача

public class QuittableTask implements Runnable {
  final int id;

  public QuittableTask(int id) {
    this.id = id;
  }

  private AtomicBoolean running = new AtomicBoolean(true);

  public void quit() {
  System.out.println("Quite " + Thread.currentThread().getName());
     running.set(false);
  }

  @Override
   public void run() {
    while (running.get()) {        
        System.out.println(Thread.currentThread().getName());
        new Nap(0.1);
    }
    System.out.print(id + " ");  
   }

Я запускаю много задач, используя

public class QuittingTasks {
  public static final int COUNT = 1500;

public static void main(String[] args) throws InterruptedException {
    ExecutorService es = Executors.newCachedThreadPool();
    List<QuittableTask> tasks =
            IntStream.range(1, COUNT)
                    .mapToObj(QuittableTask::new)
                    .peek(es::execute)
                    .collect(Collectors.toList());
    Thread.currentThread().sleep(1000);
    tasks.forEach(QuittableTask::quit);
    es.shutdown();
}

Когда я использую COUNT <10000 </strong>, все в порядке.Все темы заканчиваются.Программа завершается через 2 секунды.

Но когда я использую COUNT> 100001 , метод вполне не может быть достигнут, и моя программа переходит в бесконечный цикл.Я не понимаю такого поведения.Может ли кто-нибудь объяснить мне это?

Answer 1

Чтобы добавить ответ @ AmitBera - Executors.newCachedThreadPool() создает практически неограниченный пул потоков (см. Объяснение в разделе «Очередь» в API doc ).Таким образом, каждый новый экземпляр QuittingTask передается новому потоку и немедленно запускается в исполнение.

В какой-то момент количество QuittingTask экземпляров и новых экземпляров потока, особенно новых экземпляров потока, растет изобъем памяти.Эта «некоторая точка» чувствительна к конфигурации памяти JVM.Вы получаете 10001 в вашем.Когда я запускаю это, у меня заканчивается 4073.

Answer 2

Если вы внимательно посмотрите журналы (добавив несколько журналов в код), вы увидите это, когда программа генерирует OutOfMemoryError.И ошибка выглядит следующим образом:

java.lang.OutOfMemoryError: unable to create new native thread
    at java.lang.Thread.start0(Native Method)
    at java.lang.Thread.start(Thread.java:717)
    at java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:957)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1378)
    at java.util.stream.ReferencePipeline$11$1.accept(ReferencePipeline.java:372)
    at java.util.stream.IntPipeline$4$1.accept(IntPipeline.java:250)
    at java.util.stream.Streams$RangeIntSpliterator.forEachRemaining(Streams.java:110)
    at java.util.Spliterator$OfInt.forEachRemaining(Spliterator.java:693)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
    at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
    at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
    at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499)
    at com.java8.demo.Main.main(Main.java:58)

Это означает, что вы получаете OutOfMemory, когда вы пытаетесь вызвать quit для всей задачи, но при возникновении ошибки она никогда не выполняется, в результате вашего логического значенияфлаг никогда не устанавливается в ложь.И другой поток продолжает работать.

Вы можете заметить, что из-за нехватки памяти ВМ не может создать собственный поток. Теперь вы используете Executors.newCachedThreadPool();.Таким образом, максимальный размер пула потоков может быть Integer.MAX_VALUE, но до достижения этого значения ваша виртуальная машина не может создать новый поток.Я изменил исполнителя на FixTheradPool, например ExecutorService es = Executors.newFixedThreadPool(8);.Он отлично работает с вашим более высоким значением COUNT, так как создаст только 8 потоков.

Я поместил блок try-catch для отслеживания исключения, как показано ниже.Кроме того, я принудительно покидаю виртуальную машину, вызывая System # exit, чтобы отслеживать журналы на моей консоли.

try {
    ExecutorService es = Executors.newCachedThreadPool();
    List<QuittableTask> tasks = IntStream.range(1, COUNT).mapToObj(QuittableTask::new).peek(es::execute)
            .collect(Collectors.toList());
    Thread.currentThread().sleep(1000);
    tasks.forEach(QuittableTask::quit);
    es.shutdown();
} catch (Throwable e) {
    System.out.println("Error" + e.getMessage());
    e.printStackTrace();
    System.exit(0);
}