ExecutorService que les interruptions de tâches après un délai d'attente

Malheureusement, la solution est erronée. Il y a une sorte de bug avec ScheduledThreadPoolExecutor, a également signalé dans cette question: annulation d'une tâche de ne pas libérer les ressources de la mémoire associé à la tâche; les ressources sont publiées uniquement lorsque la tâche expire.

Si vous avez donc créer un TimeoutThreadPoolExecutor avec un assez long temps d'expiration (une utilisation typique), et de soumettre des tâches assez rapidement, vous vous retrouvez de remplissage de la mémoire - même si les tâches effectivement accomplies avec succès.

Vous pouvez voir le problème avec la suivante (très rudimentaire) programme de test:

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

Le programme épuise la mémoire disponible, bien qu'il attend pour le pondu Runnables à compléter.

J'ai bien sur ce sujet pendant un certain temps, mais malheureusement je ne pouvais pas arriver à une bonne solution.

EDIT:
J'ai découvert ce problème a été signalé comme JDK bug 6602600, et semble avoir été résolu très récemment.

Comment sur l'utilisation de l' ExecutorService.shutDownNow() méthode décrite en http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorService.html? Il semble être la solution la plus simple.

Après la tonne de temps à l'enquête

Enfin, je l'utilise invokeAll méthode de ExecutorService pour résoudre ce problème.

Qui sera strictement interrompre la tâche alors que la tâche en cours.

Ici est un exemple de

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();

Le pro est vous pouvez également soumettre ListenableFuture à la même ExecutorService.

Juste un peu modifier la première ligne de code.

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

ListeningExecutorService est l'Écoute de la fonctionnalité de ExecutorService à google de goyave projet (com.google.goyave) )

Ce à propos de cette idée alternative :

• deux ont deux exécuteurs :
  • un pour :
    • la soumission de la tâche, sans se soucier le délai d'attente de la tâche
    • ajoutant que l'Avenir abouti et le moment où elle devrait mettre fin à une structure interne
  • l'un pour l'exécution d'un poste en interne, qui est la vérification de la structure interne si certaines tâches sont timeout et si elles doivent être annulées.

Petit échantillon ici :

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