Java Executors: как я могу остановить отправленные задачи?

Question

Я отправил задание с использованием исполнителей, и мне нужно, чтобы оно было остановлено через некоторое время (например, 5 минут). Я пытался сделать так:

   for (Future<?> fut : e.invokeAll(tasks, 300, TimeUnit.SECONDS)) {
         try {
             fut.get(); 
         } catch (CancellationException ex) {
             fut.cancel(true);   
             tasks.clear();
         } catch(ExecutionException ex){
             ex.printStackTrace(); //FIXME: gestita con printstack       
         }
   }

Но я всегда получаю сообщение об ошибке: у меня есть общий Вектор, который нужно изменить задачами, а затем прочитать потоком, и даже если я остановлю всю задачу, если истечет время ожидания, я получу:

Exception in thread "Thread-1" java.util.ConcurrentModificationException

Что-то не так? Как я могу остановить отправленные задачи, которые все еще работают через 5 минут?

Answer 1

То, что вы звоните cancel() на Future, не означает, что задача остановится автоматически. Вы должны выполнить некоторую работу внутри задачи, чтобы убедиться, что она остановится:

• Используйте cancel(true), чтобы прерывание отправлялось заданию.
• Ручка InterruptedException. Если функция в вашей задаче выдает InterruptedException, убедитесь, что вы выходите изящно, как только это возможно при обнаружении исключения.
• Периодически проверять Thread.currentThread().isInterrupted(), выполняет ли задача непрерывные вычисления.

Например:

class LongTask implements Callable<Double> {
    public Double call() {

         // Sleep for a while; handle InterruptedException appropriately
         try {
             Thread.sleep(10000);
         } catch (InterruptedException ex) {
             System.out.println("Exiting gracefully!");
             return null;
         }


         // Compute for a while; check Thread.isInterrupted() periodically
         double sum = 0.0;
         for (long i = 0; i < 10000000; i++) {
             sum += 10.0
             if (Thread.currentThread().isInterrupted()) {
                 System.out.println("Exiting gracefully");
                 return null;
             }
         }

         return sum;
    } 
}

Кроме того, как уже упоминалось в других публикациях: ConcurrentModificationException может быть выброшено, даже если используется потокобезопасный класс Vector, поскольку итераторы, полученные из Vector, не являются поточно-ориентированными и, следовательно, должны быть синхронизированы. В расширенном цикле for используются итераторы, поэтому будьте внимательны:

final Vector<Double> vector = new Vector<Double>();
vector.add(1.0);
vector.add(2.0);

// Not thread safe!  If another thread modifies "vector" during the loop, then
// a ConcurrentModificationException will be thrown.
for (Double num : vector) {
    System.out.println(num);
}

// You can try this as a quick fix, but it might not be what you want:
synchronized (vector) {    // "vector" must be final
    for (Double num : vector) {
        System.out.println(num);
    }
}

Answer 2

* * * * * * * * * * * * * * * * * * * * * * * * * * * ConcurrentModificationException приходит с вашего звонка на tasks.clear(), в то время как ваши Эксперты перебирают вашу tasks Vector То, что вы можете попытаться сделать, это позвонить shutdownNow() на ваш ExecutorService

Answer 3

Поместите fut.cancel(true); в блок finally

Answer 4

fut.get () является блокирующим вызовом, даже после истечения времени ожидания вы будете блокировать, пока задача не будет выполнена. Если вы хотите остановиться как можно ближе к 5-минутной отметке, вам нужно проверить флаг прерывания, я просто рекомендую вам сделать это с помощью метода Thread.isInterrupted (), который сохраняет состояние прерывания. Если вы хотите просто немедленно остановиться и не нужно очищать какое-либо состояние, генерируйте исключение, которое будет перехвачено Future и указано вам как ExecutionException.

fut.cancel (true) ничего не делает, так как метод invokeAll () уже сделал это для вас.

Если вы не используете коллекцию "tasks" где-то еще, вам, вероятно, не нужно вызывать clear () для нее. Это не будет источником вашей проблемы, так как метод invokeAll () выполняется со списком к тому времени, когда вы вызываете clear (). Но если вам нужно начать формировать список новых задач для выполнения, я предлагаю вам сформировать новый список задач, а не использовать старый список новых задач.

К сожалению, у меня нет ответа на вашу проблему. Я не вижу достаточно информации здесь, чтобы диагностировать это. Ничто в предоставленном вами фрагменте кода не указывает на неправильное (только ненужное) использование библиотечных классов / методов. Возможно, если вы включили полную трассировку стека вместо ошибки в одну строку.

Answer 5

Наиболее распространенный случай для ConcurrentModificationException - это когда vector изменяется одновременно с итерацией. Часто это будет сделано в один поток. Вам нужно удерживать блокировку на Vector на протяжении всей итерации (и быть осторожным, чтобы не заблокировать).