Comment faire en sorte que la méthode submit () de ThreadPoolExecutor bloque s'il est saturé?

Je veux créer un ThreadPoolExecutortel que lorsqu'il a atteint sa taille maximale et que la file d'attente est pleine, la submit()méthode se bloque lors de la tentative d'ajout de nouvelles tâches. Dois-je implémenter une personnalisation RejectedExecutionHandlerpour cela ou existe-t-il un moyen de le faire à l'aide d'une bibliothèque Java standard?

Une des solutions possibles que je viens de trouver:

public class BoundedExecutor {
    private final Executor exec;
    private final Semaphore semaphore;

    public BoundedExecutor(Executor exec, int bound) {
        this.exec = exec;
        this.semaphore = new Semaphore(bound);
    }

    public void submitTask(final Runnable command)
            throws InterruptedException, RejectedExecutionException {
        semaphore.acquire();
        try {
            exec.execute(new Runnable() {
                public void run() {
                    try {
                        command.run();
                    } finally {
                        semaphore.release();
                    }
                }
            });
        } catch (RejectedExecutionException e) {
            semaphore.release();
            throw e;
        }
    }
}

Y a-t-il d'autres solutions? Je préférerais quelque chose basé sur RejectedExecutionHandlercar cela semble être un moyen standard de gérer de telles situations.

Vous pouvez utiliser ThreadPoolExecutor et un blockingQueue:

public class ImageManager {
    BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(blockQueueSize);
    RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy();
    private ExecutorService executorService =  new ThreadPoolExecutor(numOfThread, numOfThread, 
        0L, TimeUnit.MILLISECONDS, blockingQueue, rejectedExecutionHandler);

    private int downloadThumbnail(String fileListPath){
        executorService.submit(new yourRunnable());
    }
}

Vous devez utiliser le CallerRunsPolicy, qui exécute la tâche rejetée dans le thread appelant. De cette façon, il ne peut pas soumettre de nouvelles tâches à l'exécuteur jusqu'à ce que cette tâche soit terminée, à quel point il y aura des threads de pool libres ou le processus se répétera.

http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/ThreadPoolExecutor.CallerRunsPolicy.html

À partir de la documentation:

Tâches rejetées

Les nouvelles tâches soumises dans la méthode execute (java.lang.Runnable) seront rejetées lorsque l'Executor a été arrêté, et également lorsque l'Executor utilise des limites finies pour les threads maximum et la capacité de la file d'attente de travail, et est saturé. Dans les deux cas, la méthode execute appelle la méthode RejectedExecutionHandler.rejectedExecution (java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) de son RejectedExecutionHandler. Quatre stratégies de gestionnaire prédéfinies sont fournies:

1. Dans le ThreadPoolExecutor.AbortPolicy par défaut, le gestionnaire lève une exception RejectedExecutionException lors du rejet.
2. Dans ThreadPoolExecutor.CallerRunsPolicy, le thread qui appelle s'exécute exécute la tâche. Cela fournit un mécanisme de contrôle de rétroaction simple qui ralentira la vitesse à laquelle de nouvelles tâches sont soumises.
3. Dans ThreadPoolExecutor.DiscardPolicy, une tâche qui ne peut pas être exécutée est simplement supprimée.
4. Dans ThreadPoolExecutor.DiscardOldestPolicy, si l'exécuteur n'est pas arrêté, la tâche en tête de la file d'attente de travail est abandonnée, puis l'exécution est relancée (ce qui peut échouer à nouveau, provoquant la répétition de cette opération.)

Assurez-vous également d'utiliser une file d'attente limitée, telle que ArrayBlockingQueue, lors de l'appel du ThreadPoolExecutorconstructeur. Sinon, rien ne sera rejeté.

Edit: en réponse à votre commentaire, définissez la taille de l'ArrayBlockingQueue pour qu'elle soit égale à la taille maximale du pool de threads et utilisez la méthode AbortPolicy.

Edit 2: Ok, je vois où tu veux en venir. Qu'en est-il de ceci: remplacer la beforeExecute()méthode pour vérifier que getActiveCount()cela ne dépasse pas getMaximumPoolSize(), et si c'est le cas, dormir et réessayer?

Hibernate a un BlockPolicyqui est simple et peut faire ce que vous voulez:

Voir: Executors.java

/**
 * A handler for rejected tasks that will have the caller block until
 * space is available.
 */
public static class BlockPolicy implements RejectedExecutionHandler {

    /**
     * Creates a <tt>BlockPolicy</tt>.
     */
    public BlockPolicy() { }

    /**
     * Puts the Runnable to the blocking queue, effectively blocking
     * the delegating thread until space is available.
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        try {
            e.getQueue().put( r );
        }
        catch (InterruptedException e1) {
            log.error( "Work discarded, thread was interrupted while waiting for space to schedule: {}", r );
        }
    }
}

La BoundedExecutorréponse citée ci-dessus de Java Concurrency en pratique ne fonctionne correctement que si vous utilisez une file d'attente illimitée pour l'Executor, ou si la limite du sémaphore n'est pas supérieure à la taille de la file d'attente. Le sémaphore est un état partagé entre le thread soumissionnaire et les threads du pool, ce qui permet de saturer l'exécuteur même si la taille de la file d'attente <liée <= (taille de la file d'attente + taille du pool).

L'utilisation CallerRunsPolicyn'est valide que si vos tâches ne s'exécutent pas indéfiniment, auquel cas votre fil de soumission restera rejectedExecutionindéfiniment, et une mauvaise idée si vos tâches prennent beaucoup de temps à s'exécuter, car le fil de soumission ne peut pas soumettre de nouvelles tâches ou faire autre chose s'il exécute une tâche elle-même.

Si ce n'est pas acceptable, je suggère de vérifier la taille de la file d'attente limitée de l'exécuteur avant de soumettre une tâche. Si la file d'attente est pleine, attendez quelques instants avant d'essayer de soumettre à nouveau. Le débit en souffrira, mais je suggère que c'est une solution plus simple que la plupart des autres solutions proposées et vous êtes assuré qu'aucune tâche ne sera rejetée.

Je sais, c'est un hack, mais à mon avis, le hack le plus propre entre ceux proposés ici ;-)

Étant donné que ThreadPoolExecutor utilise la file d'attente de blocage "offer" au lieu de "put", nous allons remplacer le comportement de "offer" de la file de blocage:

class BlockingQueueHack<T> extends ArrayBlockingQueue<T> {

    BlockingQueueHack(int size) {
        super(size);
    }

    public boolean offer(T task) {
        try {
            this.put(task);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return true;
    }
}

ThreadPoolExecutor tp = new ThreadPoolExecutor(1, 2, 1, TimeUnit.MINUTES, new BlockingQueueHack(5));

Je l'ai testé et il semble fonctionner. La mise en œuvre d'une politique de temporisation est laissée à l'exercice du lecteur.

La classe suivante s'enroule autour d'un ThreadPoolExecutor et utilise un sémaphore pour bloquer, puis la file d'attente de travail est pleine:

public final class BlockingExecutor { 

    private final Executor executor;
    private final Semaphore semaphore;

    public BlockingExecutor(int queueSize, int corePoolSize, int maxPoolSize, int keepAliveTime, TimeUnit unit, ThreadFactory factory) {
        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
        this.executor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, unit, queue, factory);
        this.semaphore = new Semaphore(queueSize + maxPoolSize);
    }

    private void execImpl (final Runnable command) throws InterruptedException {
        semaphore.acquire();
        try {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        command.run();
                    } finally {
                        semaphore.release();
                    }
                }
            });
        } catch (RejectedExecutionException e) {
            // will never be thrown with an unbounded buffer (LinkedBlockingQueue)
            semaphore.release();
            throw e;
        }
    }

    public void execute (Runnable command) throws InterruptedException {
        execImpl(command);
    }
}

Cette classe wrapper est basée sur une solution donnée dans le livre Java Concurrency in Practice de Brian Goetz. La solution dans le livre ne prend que deux paramètres de constructeur: un Executoret une borne utilisée pour le sémaphore. Ceci est montré dans la réponse donnée par Fixpoint. Il y a un problème avec cette approche: elle peut entrer dans un état où les threads du pool sont occupés, la file d'attente est pleine, mais le sémaphore vient de publier un permis. ( semaphore.release()dans le bloc finally). Dans cet état, une nouvelle tâche peut récupérer l'autorisation qui vient d'être publiée, mais elle est rejetée car la file d'attente des tâches est pleine. Bien sûr, ce n'est pas quelque chose que vous voulez; vous souhaitez bloquer dans ce cas.

Pour résoudre ce problème, nous devons utiliser une file d'attente illimitée , comme le mentionne clairement JCiP. Le sémaphore agit comme une garde, donnant l'effet d'une taille de file d'attente virtuelle. Cela a pour effet secondaire qu'il est possible que l'unité puisse contenir des maxPoolSize + virtualQueueSize + maxPoolSizetâches. Pourquoi donc? En raison de la semaphore.release()dans le bloc finalement. Si tous les threads du pool appellent cette instruction en même temps, les maxPoolSizeautorisations sont libérées, permettant au même nombre de tâches d'entrer dans l'unité. Si nous utilisions une file d'attente limitée, elle serait toujours pleine, ce qui entraînerait le rejet d'une tâche. Maintenant, comme nous savons que cela ne se produit que lorsqu'un thread de pool est presque terminé, ce n'est pas un problème. Nous savons que le thread du pool ne se bloquera pas, donc une tâche sera bientôt retirée de la file d'attente.

Vous pouvez cependant utiliser une file d'attente limitée. Assurez-vous simplement que sa taille est égale virtualQueueSize + maxPoolSize. De plus grandes tailles sont inutiles, le sémaphore empêchera de laisser entrer plus d'éléments. Des tailles plus petites entraîneront des tâches rejetées. Le risque que les tâches soient rejetées augmente à mesure que la taille diminue. Par exemple, supposons que vous souhaitiez un exécuteur limité avec maxPoolSize = 2 et virtualQueueSize = 5. Ensuite, prenez un sémaphore avec 5 + 2 = 7 autorisations et une taille de file d'attente réelle de 5 + 2 = 7. Le nombre réel de tâches qui peuvent être dans l'unité est alors 2 + 5 + 2 = 9. Lorsque l'exécuteur est plein (5 tâches en file d'attente, 2 dans le pool de threads, donc 0 permis disponible) et TOUS les threads du pool libèrent leurs autorisations, alors exactement 2 autorisations peuvent être prises par les tâches entrantes.

Maintenant, la solution de JCiP est quelque peu lourde à utiliser car elle n'applique pas toutes ces contraintes (file d'attente illimitée, ou limitée par ces restrictions mathématiques, etc.). Je pense que cela ne sert que de bon exemple pour démontrer comment vous pouvez créer de nouvelles classes thread-safe basées sur les parties déjà disponibles, mais pas en tant que classe réutilisable à part entière. Je ne pense pas que cette dernière était l'intention de l'auteur.

vous pouvez utiliser un RejectedExecutionHandler personnalisé comme celui-ci

ThreadPoolExecutor tp= new ThreadPoolExecutor(core_size, // core size
                max_handlers, // max size 
                timeout_in_seconds, // idle timeout 
                TimeUnit.SECONDS, queue, new RejectedExecutionHandler() {
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        // This will block if the queue is full
                        try {
                            executor.getQueue().put(r);
                        } catch (InterruptedException e) {
                            System.err.println(e.getMessage());
                        }

                    }
                });

Créez votre propre file d'attente de blocage à utiliser par l'exécuteur, avec le comportement de blocage que vous recherchez, tout en renvoyant toujours la capacité restante disponible (en vous assurant que l'exécuteur n'essaiera pas de créer plus de threads que son pool de base, ou de déclencher le gestionnaire de rejet).

Je pense que cela vous permettra d'obtenir le comportement de blocage que vous recherchez. Un gestionnaire de rejet ne correspondra jamais à la facture, car cela indique que l'exécuteur ne peut pas effectuer la tâche. Ce que je pourrais imaginer, c'est que vous obtenez une forme d '«attente occupée» dans le gestionnaire. Ce n'est pas ce que vous voulez, vous voulez une file d'attente pour l'exécuteur qui bloque l'appelant ...

Pour éviter les problèmes avec la solution @FixPoint. On pourrait utiliser ListeningExecutorService et libérer le sémaphore onSuccess et onFailure dans FutureCallback.

Récemment, j'ai trouvé que cette question avait le même problème. L'OP ne le dit pas explicitement, mais nous ne voulons pas utiliser le RejectedExecutionHandlerqui exécute une tâche sur le thread de l'émetteur, car cela sous-utilisera les threads de travail si cette tâche est longue.

En lisant toutes les réponses et commentaires, en particulier la solution défectueuse avec le sémaphore ou en utilisant, afterExecutej'ai examiné de plus près le code de ThreadPoolExecutor pour voir s'il y avait un moyen de sortir. J'ai été étonné de voir qu'il y avait plus de 2000 lignes de code (commentées), dont certaines me donnent le vertige . Compte tenu de l'exigence plutôt simple que j'ai en fait - un producteur, plusieurs consommateurs, laisser le producteur bloquer quand aucun consommateur ne peut travailler - j'ai décidé de lancer ma propre solution. Ce n'est pas un ExecutorServicemais juste un Executor. Et il n'adapte pas le nombre de threads à la charge de travail, mais ne contient qu'un nombre fixe de threads, ce qui correspond également à mes besoins. Voici le code. N'hésitez pas à vous en plaindre :-)

package x;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;

/**
 * distributes {@code Runnable}s to a fixed number of threads. To keep the
 * code lean, this is not an {@code ExecutorService}. In particular there is
 * only very simple support to shut this executor down.
 */
public class ParallelExecutor implements Executor {
  // other bounded queues work as well and are useful to buffer peak loads
  private final BlockingQueue<Runnable> workQueue =
      new SynchronousQueue<Runnable>();
  private final Thread[] threads;

  /*+**********************************************************************/
  /**
   * creates the requested number of threads and starts them to wait for
   * incoming work
   */
  public ParallelExecutor(int numThreads) {
    this.threads = new Thread[numThreads];
    for(int i=0; i<numThreads; i++) {
      // could reuse the same Runner all over, but keep it simple
      Thread t = new Thread(new Runner());
      this.threads[i] = t;
      t.start();
    }
  }
  /*+**********************************************************************/
  /**
   * returns immediately without waiting for the task to be finished, but may
   * block if all worker threads are busy.
   * 
   * @throws RejectedExecutionException if we got interrupted while waiting
   *         for a free worker
   */
  @Override
  public void execute(Runnable task)  {
    try {
      workQueue.put(task);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new RejectedExecutionException("interrupt while waiting for a free "
          + "worker.", e);
    }
  }
  /*+**********************************************************************/
  /**
   * Interrupts all workers and joins them. Tasks susceptible to an interrupt
   * will preempt their work. Blocks until the last thread surrendered.
   */
  public void interruptAndJoinAll() throws InterruptedException {
    for(Thread t : threads) {
      t.interrupt();
    }
    for(Thread t : threads) {
      t.join();
    }
  }
  /*+**********************************************************************/
  private final class Runner implements Runnable {
    @Override
    public void run() {
      while (!Thread.currentThread().isInterrupted()) {
        Runnable task;
        try {
          task = workQueue.take();
        } catch (InterruptedException e) {
          // canonical handling despite exiting right away
          Thread.currentThread().interrupt(); 
          return;
        }
        try {
          task.run();
        } catch (RuntimeException e) {
          // production code to use a logging framework
          e.printStackTrace();
        }
      }
    }
  }
}

Je pense qu'il existe un moyen assez élégant de résoudre ce problème en utilisant java.util.concurrent.Semaphoreet en déléguant le comportement de Executor.newFixedThreadPool. Le nouveau service exécuteur n'exécutera une nouvelle tâche que s'il y a un thread pour le faire. Le blocage est géré par Semaphore avec un nombre d'autorisations égal au nombre de threads. Lorsqu'une tâche est terminée, elle renvoie un permis.

public class FixedThreadBlockingExecutorService extends AbstractExecutorService {

private final ExecutorService executor;
private final Semaphore blockExecution;

public FixedThreadBlockingExecutorService(int nTreads) {
    this.executor = Executors.newFixedThreadPool(nTreads);
    blockExecution = new Semaphore(nTreads);
}

@Override
public void shutdown() {
    executor.shutdown();
}

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

@Override
public boolean isShutdown() {
    return executor.isShutdown();
}

@Override
public boolean isTerminated() {
    return executor.isTerminated();
}

@Override
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
    return executor.awaitTermination(timeout, unit);
}

@Override
public void execute(Runnable command) {
    blockExecution.acquireUninterruptibly();
    executor.execute(() -> {
        try {
            command.run();
        } finally {
            blockExecution.release();
        }
    });
}

J'avais le même besoin dans le passé: une sorte de file d'attente de blocage avec une taille fixe pour chaque client adossée à un pool de threads partagé. J'ai fini par écrire mon propre type de ThreadPoolExecutor:

UserThreadPoolExecutor (file d'attente de blocage (par client) + threadpool (partagé entre tous les clients))

Voir: https://github.com/d4rxh4wx/UserThreadPoolExecutor

Chaque UserThreadPoolExecutor reçoit un nombre maximum de threads d'un ThreadPoolExecutor partagé

Chaque UserThreadPoolExecutor peut:

• soumettre une tâche à l'exécuteur du pool de threads partagés si son quota n'est pas atteint. Si son quota est atteint, le job est mis en file d'attente (blocage non consommateur en attente de CPU). Une fois qu'une de ses tâches soumises est terminée, le quota est décrémenté, permettant à une autre tâche en attente d'être soumise à ThreadPoolExecutor
• attendre que les tâches restantes soient terminées

J'ai trouvé cette politique de rejet dans le client de recherche élastique. Il bloque le fil de l'appelant sur la file d'attente de blocage. Code ci-dessous-

 static class ForceQueuePolicy implements XRejectedExecutionHandler 
 {
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) 
        {
            try 
            {
                executor.getQueue().put(r);
            } 
            catch (InterruptedException e) 
            {
                //should never happen since we never wait
                throw new EsRejectedExecutionException(e);
            }
        }

        @Override
        public long rejected() 
        {
            return 0;
        }
}

J'ai récemment eu besoin de réaliser quelque chose de similaire, mais sur un ScheduledExecutorService.

Je devais également m'assurer de gérer le retard transmis à la méthode et de m'assurer que soit la tâche est soumise pour s'exécuter au moment attendu par l'appelant, soit échoue tout simplement en lançant un fichier RejectedExecutionException.

D'autres méthodes de ScheduledThreadPoolExecutorpour exécuter ou soumettre une tâche en interne appellent #schedulequi à leur tour invoqueront toujours les méthodes remplacées.

import java.util.concurrent.*;

public class BlockingScheduler extends ScheduledThreadPoolExecutor {
    private final Semaphore maxQueueSize;

    public BlockingScheduler(int corePoolSize,
                             ThreadFactory threadFactory,
                             int maxQueueSize) {
        super(corePoolSize, threadFactory, new AbortPolicy());
        this.maxQueueSize = new Semaphore(maxQueueSize);
    }

    @Override
    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay,
                                       TimeUnit unit) {
        final long newDelayInMs = beforeSchedule(command, unit.toMillis(delay));
        return super.schedule(command, newDelayInMs, TimeUnit.MILLISECONDS);
    }

    @Override
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay,
                                           TimeUnit unit) {
        final long newDelayInMs = beforeSchedule(callable, unit.toMillis(delay));
        return super.schedule(callable, newDelayInMs, TimeUnit.MILLISECONDS);
    }

    @Override
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit) {
        final long newDelayInMs = beforeSchedule(command, unit.toMillis(initialDelay));
        return super.scheduleAtFixedRate(command, newDelayInMs, unit.toMillis(period), TimeUnit.MILLISECONDS);
    }

    @Override
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long period,
                                                     TimeUnit unit) {
        final long newDelayInMs = beforeSchedule(command, unit.toMillis(initialDelay));
        return super.scheduleWithFixedDelay(command, newDelayInMs, unit.toMillis(period), TimeUnit.MILLISECONDS);
    }

    @Override
    protected void afterExecute(Runnable runnable, Throwable t) {
        super.afterExecute(runnable, t);
        try {
            if (t == null && runnable instanceof Future<?>) {
                try {
                    ((Future<?>) runnable).get();
                } catch (CancellationException | ExecutionException e) {
                    t = e;
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt(); // ignore/reset
                }
            }
            if (t != null) {
                System.err.println(t);
            }
        } finally {
            releaseQueueUsage();
        }
    }

    private long beforeSchedule(Runnable runnable, long delay) {
        try {
            return getQueuePermitAndModifiedDelay(delay);
        } catch (InterruptedException e) {
            getRejectedExecutionHandler().rejectedExecution(runnable, this);
            return 0;
        }
    }

    private long beforeSchedule(Callable callable, long delay) {
        try {
            return getQueuePermitAndModifiedDelay(delay);
        } catch (InterruptedException e) {
            getRejectedExecutionHandler().rejectedExecution(new FutureTask(callable), this);
            return 0;
        }
    }

    private long getQueuePermitAndModifiedDelay(long delay) throws InterruptedException {
        final long beforeAcquireTimeStamp = System.currentTimeMillis();
        maxQueueSize.tryAcquire(delay, TimeUnit.MILLISECONDS);
        final long afterAcquireTimeStamp = System.currentTimeMillis();
        return afterAcquireTimeStamp - beforeAcquireTimeStamp;
    }

    private void releaseQueueUsage() {
        maxQueueSize.release();
    }
}

J'ai le code ici, j'apprécierai vos commentaires. https://github.com/AmitabhAwasthi/BlockingScheduler

Voici la solution qui semble très bien fonctionner. Il s'appelle NotifyingBlockingThreadPoolExecutor .

Programme de démonstration.

Edit: Il y a un problème avec ce code, la méthode await () est boguée. L'appel de shutdown () + awaitTermination () semble fonctionner correctement.

Je n'aime pas toujours la CallerRunsPolicy, d'autant plus qu'elle permet à la tâche rejetée de «sauter la file d'attente» et d'être exécutée avant les tâches qui ont été soumises plus tôt. De plus, l'exécution de la tâche sur le thread appelant peut prendre beaucoup plus de temps que d'attendre que le premier emplacement devienne disponible.

J'ai résolu ce problème en utilisant un RejectedExecutionHandler personnalisé, qui bloque simplement le thread appelant pendant un petit moment, puis essaie à nouveau de soumettre la tâche:

public class BlockWhenQueueFull implements RejectedExecutionHandler {

    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {

        // The pool is full. Wait, then try again.
        try {
            long waitMs = 250;
            Thread.sleep(waitMs);
        } catch (InterruptedException interruptedException) {}

        executor.execute(r);
    }
}

Cette classe peut simplement être utilisée dans l'exécuteur du pool de threads en tant que RejectedExecutinHandler comme n'importe quel autre, par exemple:

executorPool = new ThreadPoolExecutor(1, 1, 10,
                                      TimeUnit.SECONDS, new SynchronousQueue<Runnable>(),
                                      new BlockWhenQueueFull());

Le seul inconvénient que je vois est que le thread appelant peut être verrouillé un peu plus longtemps que strictement nécessaire (jusqu'à 250 ms). De plus, puisque cet exécuteur est effectivement appelé de manière récursive, de très longues attentes pour qu'un thread soit disponible (heures) peuvent entraîner un débordement de pile.

Néanmoins, j'aime personnellement cette méthode. Il est compact, facile à comprendre et fonctionne bien.