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

к сожалению, решение является ошибочным. Существует своего рода ошибка с 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сек.

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

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

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

Java native Process timeout

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

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

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

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

однако это не соответствует моим потребностям, так как некоторые операции ввода-вывода не прерываются, когда Future.cancel() называется (т. е. когда Thread.interrupted() называется).

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

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

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

небольшой пример здесь :

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