Я ищу реализацию 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();
}
}
}
java
multithreading
concurrency
executorservice
Эдвард Дейл
источник
источник
protected void beforeExecute(Thread t, Runnable r)
крючка.Ответы:
Для этого вы можете использовать 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 секунд, а затем отменит (т.е. прервет) эту конкретную задачу.
источник
beforeExecute
крючок.К сожалению, решение некорректно. Существует своего рода ошибка
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
s.Некоторое время я думал об этом, но, к сожалению, не смог придумать хорошего решения.
РЕДАКТИРОВАТЬ: Я узнал, что об этой проблеме было сообщено как об ошибке JDK 6602600 , и, похоже, она была исправлена совсем недавно.
источник
Оберните задачу в FutureTask, и вы можете указать время ожидания для FutureTask. Посмотрите на пример в моем ответе на этот вопрос,
Тайм-аут собственного процесса Java
источник
java.util.concurrent
классов, но я ищуExecutorService
реализацию.После кучи времени на изучение,
наконец, я использую
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
в проекте google guava ( com.google.guava ))источник
invokeAll
. Это очень хорошо работает. Небольшое предупреждение для всех, кто думает об этом: хотяinvokeAll
возвращает списокFuture
объектов, на самом деле это операция блокировки.Как насчет использования
ExecutorService.shutDownNow()
метода, описанного в http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorService.html ? Кажется, самое простое решение.источник
Похоже, проблема не в ошибке 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 проверенных кругов.
источник
Используя ответ Джона В., я создал реализацию, которая правильно запускает тайм-аут, когда задача начинает свое выполнение. Я даже написал для него юнит-тест :)
Однако это не соответствует моим потребностям, так как некоторые операции ввода-вывода не прерываются при
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
источник
Socket.connect
Socket.read
myThread.interrupted()
не является правильным методом прерывания, так как он ОЧИЩАЕТ флаг прерывания. ИспользуйтеmyThread.interrupt()
вместо этого, и это должно быть с сокетамиThread.interrupted()
не позволяет прервать поток. ОднакоThread.interrupt()
не прерываетjava.io
операции, он работает только сjava.nio
операциями.interrupt()
много лет, и он всегда прерывал операции java.io (а также другие методы блокировки, такие как спящий режим, соединения jdbc, блокировка очереди и т. Д.). Возможно, вы нашли класс с ошибками или какую-то JVM, в которой есть ошибкиА как насчет этой альтернативной идеи:
Небольшой образец здесь:
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; } } }
источник
проверьте, работает ли это для вас,
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; }
вы можете ограничить использование потоков из планировщика, а также установить тайм-аут для задачи.
источник