ExecutorService, который прерывает задачи после тайм-аута

95

Я ищу реализацию ExecutorService, которая может быть предоставлена ​​с таймаутом. Задачи, которые отправляются в ExecutorService, прерываются, если для их выполнения требуется больше времени, чем время ожидания. Реализовать такого зверя - не такая уж сложная задача, но мне интересно, знает ли кто-нибудь о существующей реализации.

Вот что я придумал, основываясь на некоторых из приведенных ниже обсуждений. Любые комментарии?

import java.util.List;
import java.util.concurrent.*;

public class TimeoutThreadPoolExecutor extends ThreadPoolExecutor {
    private final long timeout;
    private final TimeUnit timeoutUnit;

    private final ScheduledExecutorService timeoutExecutor = Executors.newSingleThreadScheduledExecutor();
    private final ConcurrentMap<Runnable, ScheduledFuture> runningTasks = new ConcurrentHashMap<Runnable, ScheduledFuture>();

    public TimeoutThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, long timeout, TimeUnit timeoutUnit) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        this.timeout = timeout;
        this.timeoutUnit = timeoutUnit;
    }

    public TimeoutThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, long timeout, TimeUnit timeoutUnit) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        this.timeout = timeout;
        this.timeoutUnit = timeoutUnit;
    }

    public TimeoutThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler, long timeout, TimeUnit timeoutUnit) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        this.timeout = timeout;
        this.timeoutUnit = timeoutUnit;
    }

    public TimeoutThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler, long timeout, TimeUnit timeoutUnit) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        this.timeout = timeout;
        this.timeoutUnit = timeoutUnit;
    }

    @Override
    public void shutdown() {
        timeoutExecutor.shutdown();
        super.shutdown();
    }

    @Override
    public List<Runnable> shutdownNow() {
        timeoutExecutor.shutdownNow();
        return super.shutdownNow();
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        if(timeout > 0) {
            final ScheduledFuture<?> scheduled = timeoutExecutor.schedule(new TimeoutTask(t), timeout, timeoutUnit);
            runningTasks.put(r, scheduled);
        }
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        ScheduledFuture timeoutTask = runningTasks.remove(r);
        if(timeoutTask != null) {
            timeoutTask.cancel(false);
        }
    }

    class TimeoutTask implements Runnable {
        private final Thread thread;

        public TimeoutTask(Thread thread) {
            this.thread = thread;
        }

        @Override
        public void run() {
            thread.interrupt();
        }
    }
}
Эдвард Дейл
источник
Это «время начала» тайм-аута и время отправки? Или время начала выполнения задачи?
Тим Бендер
Хороший вопрос. Когда он начнет выполняться. Предположительно с помощью protected void beforeExecute(Thread t, Runnable r)крючка.
Эдвард Дейл,
@ scompt.com, вы все еще используете это решение или оно было заменено
Пол Тейлор
@PaulTaylor Работа, в которой я реализовал это решение, заменена. :-)
Эдвард Дейл
Мне нужно именно это, за исключением: а) мне нужно, чтобы моя основная служба планировщика была пулом потоков с одним потоком службы, так как мои задачи должны выполняться строго одновременно, и б) мне нужно указать продолжительность тайм-аута для каждой задачи в время отправки задачи. Я попытался использовать это в качестве отправной точки, но расширил ScheduledThreadPoolExecutor, но я не вижу способа получить указанную продолжительность тайм-аута, которая должна быть указана во время отправки задачи, до метода beforeExecute. Любые предложения с благодарностью оценены!
Майкл Эллис

Ответы:

91

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

 ScheduledExecutorService executor = Executors.newScheduledThreadPool(2); 
 final Future handler = executor.submit(new Callable(){ ... });
 executor.schedule(new Runnable(){
     public void run(){
         handler.cancel();
     }      
 }, 10000, TimeUnit.MILLISECONDS);

Это приведет к выполнению вашего обработчика (прерывание основной функции) в течение 10 секунд, а затем отменит (т.е. прервет) эту конкретную задачу.

Джон Винт
источник
13
Интересная идея, но что, если задача завершится до истечения времени ожидания (что обычно и происходит)? Я бы предпочел не иметь кучу задач очистки, ожидающих выполнения только для того, чтобы узнать, что назначенная им задача уже выполнена. Потребуется еще один поток, отслеживающий фьючерсы по мере их завершения, чтобы удалить свои задачи очистки.
Эдвард Дейл,
4
Исполнитель планирует эту отмену только один раз. Если задача завершена, то отмена не выполняется, и работа продолжается без изменений. Требуется только один дополнительный поток, планирующий отменить задачи, и один поток для их выполнения. У вас может быть два исполнителя: один для отправки ваших основных задач, а другой для их отмены.
Джон Винт
3
Это правда, но что, если таймаут составляет 5 часов и за это время выполняется 10 тысяч задач. Я бы хотел, чтобы все эти запреты не валялись, занимая память и вызывая переключение контекста.
Эдвард Дейл,
1
@Scompt Не обязательно. Будет 10k вызовов future.cancel (), однако, если future будет завершен, отмена приведет к быстрому выходу и не будет выполнять лишнюю работу. Если вам не нужны дополнительные 10 000 вызовов отмены, это может не сработать, но объем работы, выполняемой после завершения задачи, очень мал.
Джон Винт
6
@John W .: Я только что понял еще одну проблему с вашей реализацией. Мне нужно, чтобы тайм-аут начинался, когда задача начинает выполнение, как я уже комментировал ранее. Я думаю, что единственный способ сделать это - использовать beforeExecuteкрючок.
Эдвард Дейл,
6

К сожалению, решение некорректно. Существует своего рода ошибка ScheduledThreadPoolExecutor, о которой также сообщается в этом вопросе : отмена отправленной задачи не полностью освобождает ресурсы памяти, связанные с задачей; ресурсы высвобождаются только по истечении срока действия задачи.

Поэтому, если вы создаете объект TimeoutThreadPoolExecutorс довольно длинным сроком действия (типичное использование) и отправляете задачи достаточно быстро, вы в конечном итоге заполняете память, даже если задачи на самом деле завершились успешно.

Вы можете увидеть проблему в следующей (очень грубой) тестовой программе:

public static void main(String[] args) throws InterruptedException {
    ExecutorService service = new TimeoutThreadPoolExecutor(1, 1, 10, TimeUnit.SECONDS, 
            new LinkedBlockingQueue<Runnable>(), 10, TimeUnit.MINUTES);
    //ExecutorService service = Executors.newFixedThreadPool(1);
    try {
        final AtomicInteger counter = new AtomicInteger();
        for (long i = 0; i < 10000000; i++) {
            service.submit(new Runnable() {
                @Override
                public void run() {
                    counter.incrementAndGet();
                }
            });
            if (i % 10000 == 0) {
                System.out.println(i + "/" + counter.get());
                while (i > counter.get()) {
                    Thread.sleep(10);
                }
            }
        }
    } finally {
        service.shutdown();
    }
}

Программа исчерпывает доступную память, хотя ожидает завершения порожденных Runnables.

Некоторое время я думал об этом, но, к сожалению, не смог придумать хорошего решения.

РЕДАКТИРОВАТЬ: Я узнал, что об этой проблеме было сообщено как об ошибке JDK 6602600 , и, похоже, она была исправлена ​​совсем недавно.

Флавио
источник
4

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

Тайм-аут собственного процесса Java

ZZ Coder
источник
1
Я понимаю, что есть несколько способов сделать это с помощью java.util.concurrentклассов, но я ищу ExecutorServiceреализацию.
Эдвард Дейл,
1
Если вы говорите, что хотите, чтобы ваш ExecutorService скрыл тот факт, что тайм-ауты добавляются из клиентского кода, вы можете реализовать свой собственный ExecutorService, который обертывает каждый запускаемый объект, переданный ему, с помощью FutureTask перед их выполнением.
erikprice
2

После кучи времени на изучение,
наконец, я использую invokeAllметод ExecutorServiceдля решения этой проблемы.
Это будет строго прерывать задачу во время ее выполнения.
Вот пример

ExecutorService executorService = Executors.newCachedThreadPool();

try {
    List<Callable<Object>> callables = new ArrayList<>();
    // Add your long time task (callable)
    callables.add(new VaryLongTimeTask());
    // Assign tasks for specific execution timeout (e.g. 2 sec)
    List<Future<Object>> futures = executorService.invokeAll(callables, 2000, TimeUnit.MILLISECONDS);
    for (Future<Object> future : futures) {
        // Getting result
    }
} catch (InterruptedException e) {
    e.printStackTrace();
}

executorService.shutdown();

Про это можно также представить ListenableFutureв том же самом ExecutorService.
Просто немного измените первую строку кода.

ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());

ListeningExecutorServiceэто функция прослушивания ExecutorServiceв проекте google guava ( com.google.guava ))

Джонни
источник
2
Спасибо, что указали invokeAll. Это очень хорошо работает. Небольшое предупреждение для всех, кто думает об этом: хотя invokeAllвозвращает список Futureобъектов, на самом деле это операция блокировки.
mxro
1

Как насчет использования ExecutorService.shutDownNow()метода, описанного в http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorService.html ? Кажется, самое простое решение.

Джованни Ботта
источник
8
Потому что он остановит все запланированные задачи, а не конкретную задачу, как было запрошено в вопросе
MikeL
1

Похоже, проблема не в ошибке JDK 6602600 (она была решена 22 мая 2010 г.), а в неправильном вызове сна (10) по кругу. Обратите внимание, что основной поток должен напрямую предоставлять ШАНС другим потокам для реализации своих задач, вызывая SLEEP (0) в КАЖДОЙ ветви внешнего круга. Думаю, лучше использовать Thread.yield () вместо Thread.sleep (0)

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

.......................
........................
Thread.yield();         

if (i % 1000== 0) {
System.out.println(i + "/" + counter.get()+ "/"+service.toString());
}

//                
//                while (i > counter.get()) {
//                    Thread.sleep(10);
//                } 

Корректно работает с количеством внешнего счетчика до 150 000 000 проверенных кругов.

Сергей
источник
1

Используя ответ Джона В., я создал реализацию, которая правильно запускает тайм-аут, когда задача начинает свое выполнение. Я даже написал для него юнит-тест :)

Однако это не соответствует моим потребностям, так как некоторые операции ввода-вывода не прерываются при Future.cancel()вызове (т.е. при Thread.interrupt()вызове). Некоторые примеры операций ввода-вывода, которые не могут быть прерваны при Thread.interrupt()вызове, - это Socket.connectи Socket.read(и я подозреваю, что большая часть операций ввода-вывода реализована в java.io). Все операции ввода-вывода java.nioдолжны прерываться при Thread.interrupt()вызове. Например, это касается SocketChannel.openи SocketChannel.read.

В любом случае, если кому-то интересно, я создал суть для исполнителя пула потоков, который позволяет задачам тайм-аут (если они используют прерывистые операции ...): https://gist.github.com/amanteaux/64c54a913c1ae34ad7b86db109cbc0bf

аманто
источник
Интересный код, я вставил его в свою систему, и мне любопытно, есть ли у вас примеры того, какие операции ввода-вывода не прерываются, чтобы я мог посмотреть, повлияет ли это на мою систему. Благодарность!
Дункан Кребс
@DuncanKrebs Я подробно изложил свой ответ на примере Socket.connectSocket.read
непрерывного
myThread.interrupted()не является правильным методом прерывания, так как он ОЧИЩАЕТ флаг прерывания. Используйте myThread.interrupt()вместо этого, и это должно быть с сокетами
DanielCuadra,
@DanielCuadra: Спасибо, похоже, я допустил опечатку, так как Thread.interrupted()не позволяет прервать поток. Однако Thread.interrupt()не прерывает java.ioоперации, он работает только с java.nioоперациями.
amanteaux,
Я использовал его interrupt()много лет, и он всегда прерывал операции java.io (а также другие методы блокировки, такие как спящий режим, соединения jdbc, блокировка очереди и т. Д.). Возможно, вы нашли класс с ошибками или какую-то JVM, в которой есть ошибки
DanielCuadra
0

А как насчет этой альтернативной идеи:

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

Небольшой образец здесь:

public class AlternativeExecutorService 
{

private final CopyOnWriteArrayList<ListenableFutureTask> futureQueue       = new CopyOnWriteArrayList();
private final ScheduledThreadPoolExecutor                scheduledExecutor = new ScheduledThreadPoolExecutor(1); // used for internal cleaning job
private final ListeningExecutorService                   threadExecutor    = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(5)); // used for
private ScheduledFuture scheduledFuture;
private static final long INTERNAL_JOB_CLEANUP_FREQUENCY = 1000L;

public AlternativeExecutorService()
{
    scheduledFuture = scheduledExecutor.scheduleAtFixedRate(new TimeoutManagerJob(), 0, INTERNAL_JOB_CLEANUP_FREQUENCY, TimeUnit.MILLISECONDS);
}

public void pushTask(OwnTask task)
{
    ListenableFuture<Void> future = threadExecutor.submit(task);  // -> create your Callable
    futureQueue.add(new ListenableFutureTask(future, task, getCurrentMillisecondsTime())); // -> store the time when the task should end
}

public void shutdownInternalScheduledExecutor()
{
    scheduledFuture.cancel(true);
    scheduledExecutor.shutdownNow();
}

long getCurrentMillisecondsTime()
{
    return Calendar.getInstance().get(Calendar.MILLISECOND);
}

class ListenableFutureTask
{
    private final ListenableFuture<Void> future;
    private final OwnTask                task;
    private final long                   milliSecEndTime;

    private ListenableFutureTask(ListenableFuture<Void> future, OwnTask task, long milliSecStartTime)
    {
        this.future = future;
        this.task = task;
        this.milliSecEndTime = milliSecStartTime + task.getTimeUnit().convert(task.getTimeoutDuration(), TimeUnit.MILLISECONDS);
    }

    ListenableFuture<Void> getFuture()
    {
        return future;
    }

    OwnTask getTask()
    {
        return task;
    }

    long getMilliSecEndTime()
    {
        return milliSecEndTime;
    }
}

class TimeoutManagerJob implements Runnable
{
    CopyOnWriteArrayList<ListenableFutureTask> getCopyOnWriteArrayList()
    {
        return futureQueue;
    }

    @Override
    public void run()
    {
        long currentMileSecValue = getCurrentMillisecondsTime();
        for (ListenableFutureTask futureTask : futureQueue)
        {
            consumeFuture(futureTask, currentMileSecValue);
        }
    }

    private void consumeFuture(ListenableFutureTask futureTask, long currentMileSecValue)
    {
        ListenableFuture<Void> future = futureTask.getFuture();
        boolean isTimeout = futureTask.getMilliSecEndTime() >= currentMileSecValue;
        if (isTimeout)
        {
            if (!future.isDone())
            {
                future.cancel(true);
            }
            futureQueue.remove(futureTask);
        }
    }
}

class OwnTask implements Callable<Void>
{
    private long     timeoutDuration;
    private TimeUnit timeUnit;

    OwnTask(long timeoutDuration, TimeUnit timeUnit)
    {
        this.timeoutDuration = timeoutDuration;
        this.timeUnit = timeUnit;
    }

    @Override
    public Void call() throws Exception
    {
        // do logic
        return null;
    }

    public long getTimeoutDuration()
    {
        return timeoutDuration;
    }

    public TimeUnit getTimeUnit()
    {
        return timeUnit;
    }
}
}
Ионут Месарос
источник
0

проверьте, работает ли это для вас,

    public <T,S,K,V> ResponseObject<Collection<ResponseObject<T>>> runOnScheduler(ThreadPoolExecutor threadPoolExecutor,
      int parallelismLevel, TimeUnit timeUnit, int timeToCompleteEachTask, Collection<S> collection,
      Map<K,V> context, Task<T,S,K,V> someTask){
    if(threadPoolExecutor==null){
      return ResponseObject.<Collection<ResponseObject<T>>>builder().errorCode("500").errorMessage("threadPoolExecutor can not be null").build();
    }
    if(someTask==null){
      return ResponseObject.<Collection<ResponseObject<T>>>builder().errorCode("500").errorMessage("Task can not be null").build();
    }
    if(CollectionUtils.isEmpty(collection)){
      return ResponseObject.<Collection<ResponseObject<T>>>builder().errorCode("500").errorMessage("input collection can not be empty").build();
    }

    LinkedBlockingQueue<Callable<T>> callableLinkedBlockingQueue = new LinkedBlockingQueue<>(collection.size());
    collection.forEach(value -> {
      callableLinkedBlockingQueue.offer(()->someTask.perform(value,context)); //pass some values in callable. which can be anything.
    });
    LinkedBlockingQueue<Future<T>> futures = new LinkedBlockingQueue<>();

    int count = 0;

    while(count<parallelismLevel && count < callableLinkedBlockingQueue.size()){
      Future<T> f = threadPoolExecutor.submit(callableLinkedBlockingQueue.poll());
      futures.offer(f);
      count++;
    }

    Collection<ResponseObject<T>> responseCollection = new ArrayList<>();

    while(futures.size()>0){
      Future<T> future = futures.poll();
      ResponseObject<T> responseObject = null;
        try {
          T response = future.get(timeToCompleteEachTask, timeUnit);
          responseObject = ResponseObject.<T>builder().data(response).build();
        } catch (InterruptedException e) {
          future.cancel(true);
        } catch (ExecutionException e) {
          future.cancel(true);
        } catch (TimeoutException e) {
          future.cancel(true);
        } finally {
          if (Objects.nonNull(responseObject)) {
            responseCollection.add(responseObject);
          }
          futures.remove(future);//remove this
          Callable<T> callable = getRemainingCallables(callableLinkedBlockingQueue);
          if(null!=callable){
            Future<T> f = threadPoolExecutor.submit(callable);
            futures.add(f);
          }
        }

    }
    return ResponseObject.<Collection<ResponseObject<T>>>builder().data(responseCollection).build();
  }

  private <T> Callable<T> getRemainingCallables(LinkedBlockingQueue<Callable<T>> callableLinkedBlockingQueue){
    if(callableLinkedBlockingQueue.size()>0){
      return callableLinkedBlockingQueue.poll();
    }
    return null;
  }

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

Нитиш Кумар
источник