Executor shutdownNow с заблокированными задачами?

Question

Я использую Executors для пула потоков и отправки задач. Может ли executorService.shutdownNow завершить все задачи, даже если некоторые из них могут быть заблокированы при вызове ввода/вывода в базу данных или Socket?

Answer 1

Это зависит от того, хорошо ли написаны ваши задания!

documentation говорит: «Метод shutdown() позволяет выполнять ранее отправленные задачи перед завершением, тогда как метод shutdownNow() предотвращает запуск задач ожидания и пытается остановить выполнение задач».

Однако Java не убивает потоки «из воздуха». Он пытается прерывание их. Хорошая задача будет бросать InterruptException, когда shtudownNow пытается прервать их и закончится изящно. Вы говорите о связи сокетов. Самые приличные методы блокировки клиентов будут вызывать прерывание, если они прерваны.

Примером плохой задачи может быть (довольно очевидно) для запуска потока с while(true) { readChunk(); if(endOfChunks) { break;} }. Это не делает изящную проверку прерываний! Это старое правило не использовать, пока циклы ждут, но до wait(), используя syncronized на объекте 'blocker', который может быть прерван.

Answer 2

Нет, не существует гарантий. Если вы видите документ API для ExecutorService#shutdownNow. Он говорит,

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

Если вы хотите блока, пока все задачи не будут завершены выполнение после запроса завершения работы используйте ExecutorService#awaitTermination.

Answer 3

Проще говоря: вы не можете положиться на это. ExecutorService просто interrupts выполняемые задачи; это зависит от выполнения задач, если они действительно отменяют свои усилия. Некоторые операции ввода/вывода могут (и будут) прерываться, особенно java.nio, но java.io, скорее всего, не прерывается. См. What does java.lang.Thread.interrupt() do? для получения более подробного объяснения.

Answer 4

Если не удается обработать прерывания (java.io), требуется нестандартная логика выключения.

Мое решение для инкапсуляции этой проблемы объединяет примеры «TrackingExecutorService» и «SocketUsingTask» из «Java Concurrency In Practice».

• Определить интерфейс «Shutdownable»
• Продлить ThreadPoolExecutor для отслеживания работы, представленные задач, которые реализуют интерфейс «Shutdownable»
• переопределения shutdownNow ThreadPoolExecutor, чтобы вызвать нестандартные отключения логики через интерфейс «Shutdownable»

    public interface Shutdownable { 
     public void shutdown(); 
    } 

    public class ShutdowningExecutor extends ThreadPoolExecutor{ 
     private final Set runningShutdownables 
      = Collections.synchronizedSet(new HashSet()); 

     @Override 
     protected RunnableFuture newTaskFor(final Callable callable){ 
      if (callable instanceof Shutdownable) { 
      runningShutdownables.add((Shutdownable) callable);   
      return super.newTaskFor(new Callable(){ 
        @Override 
        public T call() throws Exception { 
       T t = callable.call(); 
       runningShutdownables.remove((Shutdownable) callable); 
         return t; 
        } 
       }); 
      } else 
       return super.newTaskFor(callable); 
     } 

     public void shutdownAll() { 
      for(Shutdownable shutdownable : runningShutdownables) { 
       shutdownable.shutdown(); 
      } 
     } 

     @Override 
     public List shutdownNow(){ 
      shutdownAll(); 
      return super.shutdownNow(); 
     } 
    } 

    public abstract class ShutdownableViaCloseable implements Shutdownable{ 
     private Closeable closeable; 

     protected synchronized void setCloseable(Closeable c) { closeable = c; } 

     public synchronized void shutdown() { 
      try { 
       if (closeable != null) 
      closeable.close(); 
      } catch (IOException ignored) { } 
     } 
    } 

    public class MySocketTask extends ShutdownableViaCloseable implements Callable { 
     public MySocketTask(Socket s) { 
      setCloseable(s); 
      //constructor stuff 
     } 

     public Void call() { 
      try (Socket socket = this.socket) { 
       while(!socket.isClosed) { 
        //do stuff 
       } 
      } 
     } 
    }