Advanced   Java   Services

Executors.newSingleThreadExecutor()

Auf dieser und den nächsten Seiten folgt eine kleine Übersicht über die Services die die Factoryklasse Executors liefert.

Wie der Name schon sagt, arbeitet ein SingleThreadExecutor mit einem einzigen Thread. Da alle von der Factory Executors erzeugten Objekte vom Typ Executorservice sind, lassen sich die in diesem Interface vereinbarten Methoden aufrufen. Alle übergebenen Tasks werden sequenziell von einem einzigen Thread abgearbeitet. das folgende Beispiel zeigt dies.

Das Beispiel arbeitet mit drei Callables, die ihren Namen und zwei Zeiten ausgeben, der Anfangszeit und der Endzeit. Die Zeitdauer wird über Math.random() zufällig variiert.

private static void startCallablesWithInvokeAll()
{
  Callable<String> ca1 =
      () -> { System.out.println("1 " + Thread.currentThread().getName() + " start = " + LocalTime.now());
            int timeout = 10 + (int)(Math.random()*200);
            TimeUnit.MILLISECONDS.sleep(timeout);
            return "1 " + Thread.currentThread().getName() + " end   = " + LocalTime.now();
          } ;

  Callable<String> ca2 =
      () -> { System.out.println("2 " + Thread.currentThread().getName() + " start = " + LocalTime.now());
            int timeout = 10 + (int)(Math.random()*200);
            TimeUnit.MILLISECONDS.sleep(timeout);
            return "2 " + Thread.currentThread().getName() + " end   = " + LocalTime.now();
          } ;

  Callable<String> ca3 =
      () -> { System.out.println("3 " + Thread.currentThread().getName() + " start = " + LocalTime.now());
            int timeout = 10 + (int)(Math.random()*200);
            TimeUnit.MILLISECONDS.sleep(timeout);
            return "3 " + Thread.currentThread().getName() + " end   = " + LocalTime.now();
          } ;

  List<Callable<String>> list = new ArrayList<>();
  list.add(ca1);
  list.add(ca2);
  list.add(ca3);
  List<Future<String>> futureList = null;
  ExecutorService ste = Executors.newSingleThreadExecutor();
  try
  {
    futureList = ste.invokeAll(list);
    ste.shutdown();
    for(Future<String> fut : futureList)
      System.out.println(fut.get()); // ExecutionException
  }
  catch(InterruptedException | ExecutionException ex)
  {
    System.out.println(ex);
  }
}

Eine mögliche Ausgabe. Man sieht, daß alle Callables von einem einzigen Thread ausgeführt werden und strikt sequentiell abgearbeitet werden.

1 pool-1-thread-1 start = 11:16:12.407
2 pool-1-thread-1 start = 11:16:12.579
3 pool-1-thread-1 start = 11:16:12.778
1 pool-1-thread-1 end   = 11:16:12.579
2 pool-1-thread-1 end   = 11:16:12.778
3 pool-1-thread-1 end   = 11:16:12.865

Mit ein wenig Reflection erfährt man, daß die Methode newSingleThreadExecutor() ein Objekt des Typs java.util.concurrent.Executors$FinalizableDelegatedExecutorService liefert. Die Klasse ist eine statische innere Klasse in Executors. FinalizableDelegatedExecutorService wiederum leitet sich von DelegatedExecutorService ab und letzterer leitet sich schließlich von AbstractExecutorService ab.

AbstractExecutorService

Executors$DelegatedExecutorService

Executors$FinalizableDelegatedExecutorService

Ein Blick in den Quellcode zeigt, daß der SingleThreadExecutor im Wesentlichen ein spezieller ThreadPoolExecutor ist, der seine Runnables in einer LinkedBlockingQueue speichert.

public static ExecutorService newSingleThreadExecutor()
{
   return new FinalizableDelegatedExecutorService(
               new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()) );
}

Die statische innere Klasse Executors$FinalizableDelegatedExecutorService stellt sicher, daß spätestens bei der Zerstörung des Objekts der Executor heruntergefahren un damit beendet wird. shutDown() sollte aber sofort aufgerufen werden nachdem der Executor seine Arbeit gemacht hat.

static class FinalizableDelegatedExecutorService extends DelegatedExecutorService
{
   FinalizableDelegatedExecutorService(ExecutorService executor)
   {
      super(executor);
   }
   protected void finalize()
   {
      super.shutdown();
   }
}

Die statische innere Klasse Executors$DelegatedExecutorService bringt von dem im Konstruktor (siehe oben) übergebenen ThreadPoolExecutor nur die Methoden des Interfaces nach außen (daher ihr Name).

/**
* A wrapper class that exposes only the ExecutorService methods
* of an ExecutorService implementation.
*/
static class DelegatedExecutorService extends AbstractExecutorService
{
  private final ExecutorService e;

  DelegatedExecutorService(ExecutorService executor) { e = executor; }

  public void execute(Runnable command) { e.execute(command); }
  public void shutdown() { e.shutdown(); }
  public List<Runnable> shutdownNow() { return e.shutdownNow(); }
  public boolean isShutdown() { return e.isShutdown(); }
  public boolean isTerminated() { return e.isTerminated(); }

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

  public Future<?> submit(Runnable task) { return e.submit(task);
  }

  public <T> Future<T> submit(Callable<T> task) { return e.submit(task); }

  public <T> Future<T> submit(Runnable task, T result) { return e.submit(task, result); }

  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
      return e.invokeAll(tasks);
  }

  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
      return e.invokeAll(tasks, timeout, unit);
  }

  public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
      return e.invokeAny(tasks);
  }

  public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
      throws InterruptedException, ExecutionException, TimeoutException
  {
      return e.invokeAny(tasks, timeout, unit);
  }
}

Executors.newSingleThreadExecutor() liefert einen ExecutorService, der einen eingeschränkten ThreadPoolExecutor liefert, der nur die Methoden des Interfaces ExecutorService zeigt und dessen Poolgröße 1 ist. Er arbeitet mit einer LinkedBlockingQueue<Runnable>. Da er auf einem ThreadPoolExecutor basiert verwendet er die DefaultThreadFactory.