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

Ответ 1

Для этого вы можете использовать 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 секунд, затем отменит (т.е. прерывает) эту конкретную задачу.

Ответ 2

К сожалению, решение ошибочно. Существует некоторая ошибка с 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();
    }
}

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

Я хоть об этом некоторое время, но, к сожалению, я не смог найти хорошее решение.

EDIT: Я узнал, что этот вопрос был опубликован как ошибка JDK 6602600 и, похоже, был исправлен совсем недавно.

Ответ 3

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

java собственный тайм-аут процесса

Ответ 4

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

Ответ 5

Кажется, проблема не в ошибке JDK 6602600 (она была решена в 2010-05-22), но в неправильный вызов сна (10) по кругу. Добавим, что главная Тема должна дать прямо CHANCE для других потоков, чтобы реализовать свои задачи, вызвав 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 проверенных кругов.

Ответ 6

После тонны времени для обследования,
Наконец, я использую метод 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))

Ответ 7

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

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

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

Ответ 8

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

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

Маленький образец находится здесь:

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;
    }
}
}

Ответ 9

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

    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;
  }

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