8. Współbieżna Java: zadania i wykonawcy

Głównym celem tego (i następnego) wykładu jest przedstawienie nowych środków programowania współbieżnego, dostępnych w pakiecie java.util.concurrent. Rozpoczynamy od przypomnienia podstawowych pojęć, po czym zajmiemy się podstawowymi nowymi koncepcjami "konkurencyjnej Javy".  Tutaj uwagę skupiamy na zadaniach i wykonawcach.

1. Programowanie współbieżne - przypomnienie podstawowych pojęć

2. Pakiet java.util.concurrent z lotu ptaka

• umożliwia separację zadań i wątków,
• pozwala na anulowanie zadań,
• dostarcza puli wątków prowadzonych przez Wykonawców,
• zawiera bardziej efektywne przy "high contention" - bazowe synchronizatory (Lock),
• umożliwia łatwą realizację read/write locks,
• udostępnia nowe synchronizatory wyższego poziomu (semafory, bariery, zasuwy),
• pozwala na użycie blokujących kolejek (łatwa koordynacja zadań),
• dostarcza konkurencyjnych (efektywnych przy dużej konkurencji) kolekcji,
• definiuje atomiki (dane podstawowych typów, bezpiecznie dostępne współbieżnie).

3. Separacja zadań od sposobu ich wykonania

import java.net.*;
import java.io.*;

public class SomeServer  {

  private ServerSocket ss = null;
  private volatile boolean serverRunning = true;

  public SomeServer(ServerSocket ss) {
    this.ss = ss;
    System.out.println("Server started");
    serviceConnections();
  }


  private void serviceConnections() {
    while (serverRunning) {
      try {
        final Socket conn = ss.accept();
        System.out.println("Connection established");

        Runnable serviceCode = new Runnable() {
           public void run() {
              serviceRequests(conn);
           }
        }

        new Thread(serviceCode).start();

      } catch (Exception exc) {
          exc.printStackTrace();
      }
    }
    try { ss.close(); } catch (Exception exc) { //... }
  }

//...

}

public interface Executor {
  void execute(Runnable);
}

  Executor executor = ....;

  private void serviceConnections() {
    while (serverRunning) {
      try {
        final Socket conn = ss.accept();
        System.out.println("Connection established");

        Runnable serviceCode = new Runnable() {
           public void run() {
              serviceRequests(conn);
           }
        }

        executor.execute(serviceCode);

      } catch (Exception exc) {
          exc.printStackTrace();
      }
    }
    try { ss.close(); } catch (Exception exc) { //... }
  }

• Wykonawca uruchamiający podane mu zadania w jednym wątku (po kolei) (Executors.newSingleThreadExecutor()),
• Wykonawca, prowadzący pulę wątków o zadanych maksymalnych rozmiarach (Executors.newFixedThreadPool()),
• Wykonawca, prowadzący pulę wątków o dynamicznych rozmiarach
• (Executors.newCachedThreadPool()),
• Wykonawcy zarządzający tworzeniem i wykonaniem wątków w określonym czasie lub z określoną periodycznością (Executors.newScheduled....())

import java.util.concurrent.*;


class Task implements Runnable {

  private String name;

  public Task(String name) {
    this.name = name;
  }

  public void run() {
    for (int i=1; i <= 4; i++) {
      System.out.println(name + " " + i);
      Thread.yield();
    }
  }
}


public class Wykonawca {

  public static void main(String[] args) {
    Executor exec = Executors.newFixedThreadPool(2);
    for (int i=1; i<=4; i++) {
      exec.execute(new Task("Task " + i));
    }
  }
}

  public static void main(String[] args) {
    ExecutorService exec = Executors.newFixedThreadPool(2);
    for (int i=1; i<=4; i++) {
      exec.execute(new Task("Task " + i));
    }
    Thread.yield();
    exec.shutdown();

    try {
      exec.execute(new Task("Task after shutdown"));
    } catch (RejectedExecutionException  exc) {
        exc.printStackTrace();
    }
    try {
      exec.awaitTermination(5, TimeUnit.SECONDS);
    } catch(InterruptedException exc) { exc.printStackTrace(); }
    System.out.println("Terminated: " + exec.isTerminated());

  }

    try {
      Thread.sleep(1000);
    } catch(Exception exc) {}

    exec.shutdownNow();

4. Kończenie zadań przez interrupt()

  public void run() {
    for (byte i=1; i <= 128 ; i++) {
      if (Thread.currentThread().isInterrupted()) return;
      System.out.println(name + " " + i);
      Thread.yield();
    }
  }

class Interruptible  {

  Lock lock = new ReentrantLock();

  Runnable task1 = new Runnable() {
     public void run() {
       System.out.println("Task 1 begins");
       try {
         lock.tryLock(1000, TimeUnit.SECONDS);  // próba zamknięcia rygla (czeka na wolny rygiel lub 1000 sekund)
         System.out.println("Task 1 entered");
       } catch(InterruptedException exc) {
           System.out.println("Task 1 interrupted");
       }
       System.out.println("Task 1 stopped");
     }
  };

  Runnable task2 = new Runnable() {
    public void run() {
      System.out.println("Task 2 begins");
      for (int i=1; i <= 600; i++) {
        if (Thread.currentThread().isInterrupted()) break;
        // jakieś obliczenie
        if (Thread.currentThread().isInterrupted()) break;  // chcemy przerwać możliwie najszybciej
        try {                                               // sleep() jest przerywane pzrez interrupt()!
          Thread.sleep(1000);
        } catch (InterruptedException exc) { break; }
      }
      System.out.println("Task 2 stopped");
    }
  };


  Runnable task3 = new Runnable() {
    Scanner scan = new Scanner(  // musimy miec InterruptibleChannel, aby móc przerwać czekanie na wejściu
                    new FileInputStream(FileDescriptor.in).getChannel(), "Cp852");
    public void run() {
      System.out.println("Task 3 begins");
      System.out.print(">>");
      while (scan.hasNextLine()) {
        try {
          String s = scan.nextLine();
          System.out.print('\n'+s + "\n>>");
        } catch (Exception exc) {
            // Uwaga: scanner nie zgłasza wyjątków, ale przerywa dzialanie
            exc.printStackTrace();
            break;
        }
      }
      System.out.println("Task 3 stopped - " + scan.ioException());  // jaki wyjątek go przerwał?
    }
  };

  Interruptible() {
    ExecutorService exec = Executors.newCachedThreadPool();

    exec.execute(new Runnable() {        // wątek zamyka rygiel
                    public void run() {
                      lock.lock();
                    }
                 }
     );
    exec.execute(task1);
    exec.execute(task2);
    exec.execute(task3);
    JOptionPane.showMessageDialog(null, "Press Ok to stop all tasks");
    exec.shutdownNow();
  }

}

5. Odbiór wyników zadań. Zadania "z prawdziwego zdarzenia".

public interface Callable<V> {
  V call() throws Exception;
}

• zwrot wyniku,
• możliwość zgłoszenia wyjątku kontrolowanego.

/*
* @since 1.5
* @author Doug Lea
*/
public class FutureTask<V> implements Future<V>, Runnable {
    // .....
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        sync = new Sync(callable);
    }

    public FutureTask(Runnable runnable, V result) {
        sync = new Sync(Executors.callable(runnable, result));
    }
    // .....
    private final class Sync extends AbstractQueuedSynchronizer {
        // .....
        private final Callable<V> callable;
        // .....
        Sync(Callable<V> callable) {
            this.callable = callable;
        }
        // .....
    }
}

Runnable r = new Runnable() {
  public void run() {
    // ...
  }
};

Callable<String> c = new Callable<String>() {
   public String call() throws Exception {
     String result;
     // ...
     return result;
   }
};


FutureTask<Boolean> task1 = new FutureTask<Boolean>(r, true);
FutureTask<Object> task2 = new FutureTask<?>(Executors.callable(r));
FutureTask<String> task3 = new FutureTask<String>(c);

Executor exec = Executors.new....();  

exec.execute(task1);
exec.execute(task1);
exec.execute(task2);

ExecutorService exec = Executors.new....();  

Future<Boolean> future1 = exec.submit(task1, true);
Future<Object> future2 = exec.submit(task1);
Future<String> future3 = exec.submit(task2);

/*
 * @author Doug Lea
 */
public abstract class AbstractExecutorService implements ExecutorService {

    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        FutureTask<Object> ftask = new FutureTask<Object>(task, null);
        execute(ftask);
        return ftask;
    }

    public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        FutureTask<T> ftask = new FutureTask<T>(task, result);
        execute(ftask);
        return ftask;
    }

    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        FutureTask<T> ftask = new FutureTask<T>(task);
        execute(ftask);
        return ftask;
    }

// ...
}

import javax.swing.*;
import java.awt.event.*;
import java.util.List;
import java.util.concurrent.*;
import java.lang.reflect.*;

public class Exec1 extends JFrame implements ActionListener {

  int k = 0;
  int n = 15;
  JTextArea ta = new JTextArea(40,20);

  Exec1() {
    add(new JScrollPane(ta));
    JPanel p = new JPanel();
    JButton b = new JButton("Start");
    b.addActionListener(this);
    p.add(b);
    b = new JButton("Stop current");
    b.setActionCommand("Stop");
    b.addActionListener(this);
    p.add(b);
    b = new JButton("Curent result");
    b.setActionCommand("Result");
    b.addActionListener(this);
    p.add(b);
    b = new JButton("Shutdown");
    b.addActionListener(this);
    p.add(b);
    b = new JButton("ShutdownNow");
    b.addActionListener(this);
    p.add(b);
    add(p, "South");
    setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
    pack();
    setVisible(true);
  }

  public void actionPerformed(ActionEvent e)  {
    String cmd = e.getActionCommand();
    try {
      Method m = this.getClass().getDeclaredMethod("task"+cmd);
      m.invoke(this);
    } catch(Exception exc) { exc.printStackTrace(); }
  }


  class SumTask implements Callable<Integer> {

    private int taskNum,
                limit;

    public SumTask(int taskNum, int limit) {
      this.taskNum = taskNum;
      this.limit = limit;
    }

    public Integer call() throws Exception {
      int sum = 0;
      for (int i = 1; i <= limit; i++) {
        if (Thread.currentThread().isInterrupted()) return null;
        sum+=i;
        ta.append("Task " + taskNum + " part result = " + sum + '\n');
        Thread.sleep(1000);
      }
      return sum;
    }
  };

  Future<Integer> task;

  //ExecutorService exec = Executors.newSingleThreadExecutor();
  ExecutorService exec = Executors.newFixedThreadPool(3);

  public void taskStart() {
    try {
      task = exec.submit(new SumTask(++k, 15));
    } catch(RejectedExecutionException exc) {
        ta.append("Execution rejected\n");
        return;
    }
    ta.append("Task " + k + " submitted\n");
  }

  public void taskResult() {
    String msg = "";
    if (task.isCancelled()) msg = "Task cancelled.";
    else if (task.isDone()) {
      try {
        msg = "Ready. Result = " + task.get();
      } catch(Exception exc) {
          msg = exc.getMessage();
      }
    }
    else msg = "Task is running or waiting for execution";
    JOptionPane.showMessageDialog(null, msg);
  }

  public void taskStop() {
    task.cancel(true);
  }

  public void taskShutdown() {
    exec.shutdown();
    ta.append("Executor shutdown\n");
  }

  public void taskShutdownNow() {
    List<Runnable> awaiting = exec.shutdownNow();
    ta.append("Eeecutor shutdown now - awaiting tasks:\n");
    for (Runnable r : awaiting) {
      ta.append(r.getClass().getName()+'\n');
    }

 }


  public static void main(String[] args) {
     new Exec1();
  }

}

1. warto zaobserwować, że w call() nie musimy obsługiwać wyjątku InterruptedException przy sleep (throws Exception)
2. jednak pisząc kody zadań nie możemy pozbyć się  myślenia w kategoriach wątków (isInterrupted(), sleep)
3. shutdownNow zwraca listę zadań oczekujących na wykonanie w momencie zamknięcia serwisu - to są ogólnie Runnable, ale ponieważ zostały przekazane do wykonania jako FutureTasks - prawdziwą klasą jest FutureTask.

6. Obsługa wyników zadań za pomocą odwołań zwrotnych

Callable<Integer> callable = ...;
ExecutorService exec = Executors.new...();
exec.execute( new FutureTask<Integer>(callable) {
    public void done() {
      // kod będzie wywołany, gdy zadanie się zakończy lub zostanie anulowane
    }
});

import javax.swing.*;
import java.awt.*;
import java.awt.event.*;
import java.util.concurrent.*;


class  FutureTaskCallback<V> extends  FutureTask<V> {

  public FutureTaskCallback(Callable<V> callable) {
    super(callable);
  }

  public void done() {
    String result = "Wynik: ";
    if (isCancelled()) result += "Cancelled.";
    else try {
      result += get();
    } catch(Exception exc) {
        result += exc.toString();
    }
    JOptionPane.showMessageDialog(null, result);
  }

}


public class SimpleCallback extends JFrame {

  JTextField input = new JTextField(40),
             progress = new JTextField(40);

  String toReverse;

  Callable<String> reverseTask = new Callable<String>() {
    public String call() throws Exception {
      Thread t = Thread.currentThread();
      if (toReverse == null || toReverse.trim().equals(""))
        throw new IllegalArgumentException("Set string to reverse");
      if (t.isInterrupted()) return null;
      char[] org = toReverse.toCharArray();
      StringBuffer out = new StringBuffer();
      if (t.isInterrupted()) return null;
      for (int i = org.length-1; i>=0; i--) {
        Thread.sleep(500);
        out.append(org[i]);
        if (t.isInterrupted()) return null;
        progress.setText(out.toString());
        if (t.isInterrupted()) return null;
      }
      return out.toString();
    }
  };

  ExecutorService exec = Executors.newSingleThreadExecutor();
  FutureTaskCallback<String> ft;


  public SimpleCallback() {
    Font f = new Font("Dialog", Font.PLAIN, 16);
    input.setFont(f);
    progress.setFont(f);
    JPanel p = new JPanel();
    JButton b = new JButton("Start");
    b.addActionListener(new ActionListener() {
      public void actionPerformed(ActionEvent e) {
        ft = new FutureTaskCallback<String>(reverseTask);
        toReverse = input.getText();
        exec.execute(ft);
      }
    });
    p.add(b);
    b = new JButton("Stop");
    b.addActionListener(new ActionListener() {
      public void actionPerformed(ActionEvent e) {
        if (ft != null) ft.cancel(true);
      }
    });
    p.add(b);
    Container cp = getContentPane();
    cp.setLayout(new BoxLayout(cp, BoxLayout.Y_AXIS));
    add(input);
    add(p);
    add(progress);
    setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
    pack();
    setVisible(true);
  }





  public static void main(String[] args) {
     new SimpleCallback();
  }

}

import java.lang.reflect.Method;
import java.util.concurrent.*;

public class Ftask<V> extends FutureTask<V> {

  private Method handlerMethod;
  private Object handlerObject;

  public Ftask(Callable<V> callable, Object handler, String mname) throws Exception {
    super(callable);
    handlerObject = handler;
    handlerMethod = handler.getClass().getDeclaredMethod(mname, Object.class, Exception.class);
  }

  public void done() {
    V result = null;
    try {
      result = (V) this.get();
    } catch(Exception exc) {
      try {
        handlerMethod.invoke(handlerObject, null, exc);
      } catch(Exception ex) {
          ex.printStackTrace();
      }
      return;
    }
    try {
      handlerMethod.invoke(handlerObject, result, null);
    } catch(Exception exc) {
      exc.printStackTrace();
  }
}

public abstract class AbstractTask<V> implements Callable<V> {

  private String   name;
  private FutureTask<V> task;

  public AbstractTask(String name, Object resultHandler, String handlerMethodName) throws Exception {
    this.name = name;
    task = new Ftask<V>(this, resultHandler, handlerMethodName);
  }

  public FutureTask<V> getTask() { return task; }
  public String getName() { return name; }

}

class ResultHandler<V> {

  int x = 400, y = 50;
  public void handleResult(V result, Exception exc) {
    String msg;
    if (exc != null) msg = exc.toString();
    else msg = "Wynik = " + result;
    JFrame f = new JFrame("Task results");
    JLabel lab = new JLabel("        " + msg);
    f.add(lab);
    f.setBounds(x, y, 300, 100);
    f.setDefaultCloseOperation(JFrame.DISPOSE_ON_CLOSE);
    x+=50;
    y+=50;
    f.setVisible(true);

  }
}

  class SumTaskA extends AbstractTask<Integer> {

    public SumTaskA(String taskName, ResultHandler h, String handlerMethod )
                 throws Exception
    {
      super(taskName, h, handlerMethod);
    }

    public Integer call() throws Exception {
      Future<Integer> task = this.getTask();
      int sum = 0;
      if (task.isCancelled()) return null;
      for (int i = 1; i <= 10; i++) {
        if (task.isCancelled()) break;
        sum+=i;
        append(getName() + " part result = " + sum + '\n');
        Thread.sleep(1000);
      }
      return sum;
    }
  };


  ExecutorService exec = Executors.newFixedThreadPool(3);
  ResultHandler<Integer> handler = new ResultHandler<Integer>();
  SumTaskA task = new SumTaskA("Task 1", handler, "handleResults")
  exec.execute(task.getTask());

import java.util.List;
import java.util.concurrent.*;

public class RepeatedTaskTest {

  FutureTask<Object> task = new FutureTask<Object>( new Callable<Object>() {
                                      public Object call() {
                                        for (int i= 1; i <= 3; i++) {
                                          System.out.println(i);

                                        }
                                        return null;
                                      }
                                   }) {
           public void done() {
             System.out.println("Done.");
           }
  };

  public RepeatedTaskTest() {
    ExecutorService exec = Executors.newSingleThreadExecutor();
    System.out.println("Starting first task");
    exec.execute(task);
    try {
      System.out.println("Starting second task");
      exec.execute(task);
      Thread.sleep(5000);
    } catch(Exception exc) {
      exc.printStackTrace();
    }
    System.out.println("Executor shutdown now");
    List<Runnable> awaiting = exec.shutdownNow();
    for (Runnable r : awaiting) {
      System.out.println(r.getClass().getName()+'\n');
    }
    try {
      exec.awaitTermination(5, TimeUnit.SECONDS);
    } catch(InterruptedException exc) { exc.printStackTrace(); }

  }


  public static void main(String[] args) {
    new RepeatedTaskTest();
  }  

7. Wykonawcy a pule wątków

import java.util.concurrent.*;
import java.util.*;

class Eval implements Callable<Integer> {

  Integer num;

  public Eval(int n) {
    num = n;
  }

  public Integer call() throws Exception {
    Thread.sleep(1000);
    return num;
  }

}

public class InvokeTest {

  public static int sum(ExecutorService exec, List<Callable<Integer>> tasks) throws Exception {
    long start = System.currentTimeMillis();
    System.out.println("Start");
    List<Future<Integer>> results =  exec.invokeAll(tasks);  // InterruptedException
    long elapsed = System.currentTimeMillis() - start;
    System.out.println("End after " + elapsed/1000 + " sec.");
    int sum = 0;
    for (Future<Integer> r : results) sum += r.get();
    return sum;
  }

  public static void main(String[] args) {
    List<Callable<Integer>> taskList = new ArrayList<Callable<Integer>>();
    ExecutorService exec = Executors.newFixedThreadPool(10);
    for (int i=1; i <=5; i++) {
      Callable<Integer> task = new Eval(i);
      taskList.add(task);
    }
    try {
      int result = sum(exec, taskList);
      System.out.println("Wynik: " + result);
    } catch(Exception exc) { exc.printStackTrace(); }
  }

}

• poolCoreSize,
• poolMaximumSize,
• keepAliveTime,
• threadFactory,
• workQueue

• gdy n < poolCoreSize, to jest tworzony nowy wątek, nawet wtedy, gdy w puli są nieaktywne wątki (a zatem zadanie,
• gdy poolCoreSize < n < maximumPoolSize, to nowy wątek jest tworzony tylko wtedy, gdy kolejka zadań jest pełna, jeśli nie jest wypełniona, to zadanie jest dodawane do kolejki, a wątek nie jest tworzony (bo być może będzie można ponownie użyć innego),
• wątki od 1 do poolCoreSize są normalnie tworzone tylko "na życzenie" (gdy przychodzi zadanie), można jednak utworzyć  te wątki metodą prestartCoreThread lub prestartAllCoreThreads,
• keepAliveTime określa okres czasu utrzymywania w puli nieaktywnych wątków: nieaktywne wątki są usuwane z puli, ale tylko wtedy, gdy n > corePoolSize (w przeciwnym razie nieaktywne nie są usuwane),
• nowe wątki tworzone są metodami fabrycznymi ThreadFactory; normalnie DefaultThreadFactory, ale można podać własną,
• workQuee jest używane jako kolejka zadań do wykonania (czekających!) i musi implementować interfejs BlockingQueue<Runnable>; standardowo jest to klasa LinkedBlockingQueue<Runnable>.

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }

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

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

    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

    public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>(),
                                      threadFactory);

import java.awt.*;
import java.awt.event.*;
import java.util.concurrent.*;
import javax.swing.*;
import java.lang.reflect.*;


class Code implements Callable<Integer> {

  String name;
  int num;

  public Code(String n, int sleep) {
     name = n;
     num = sleep;
  }

  public Integer call() throws Exception {
    Thread.sleep(num*1000);
    return num;
  }

  public String getName() { return name; }
}


class Task extends FutureTask<Integer> {
  Code code;

  public Task(Code c) {
    super(c);
    code = c;
  }

  public String getName() {
    return code.getName();
  }
}



public class ThreadPoolMethods extends JFrame implements ActionListener {

  ThreadPoolExecutor exec = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);

  public ThreadPoolMethods() {
    setLayout(new FlowLayout());
    JButton b = new JButton("Start");
    b.addActionListener(this);
    add(b);
    b = new JButton("Show");
    b.addActionListener(this);
    add(b);
    pack();
    setVisible(true);
  }

  public void actionPerformed(ActionEvent e) {
    String cmd = e.getActionCommand();
    if (cmd.equals("Start")) {
      int sleep = 5, step = 3;
      for (int k=1; k <= 10; k++) {
        sleep += step;
        exec.execute(new Task(new Code("Task " + k, sleep)));
      }
      System.out.println("All tasks submitted");
    }
    else {
      try {
        printStat();
      } catch(Exception exc) { exc.printStackTrace(); }
      BlockingQueue<Runnable> que = exec.getQueue();
      System.out.println("Que size " + que.size());
      for (Runnable r : que) {
        Task t = (Task) r;
        System.out.println(t.getName());
      }
    }
  }

  void printStat() throws Exception {
    Method[] mets = ThreadPoolExecutor.class.getDeclaredMethods();
    for (Method m : mets) {
      String name = m.getName();
      if (name.startsWith("get")) {
        Object res = null;
        try {
          res = m.invoke(exec);
        } catch(Exception exc) {
          System.out.println("Unable to call " + name);
        }
        System.out.println(name + " = " + res);
      }
    }
  }

  public static void main(String[] args) {
    new ThreadPoolMethods();
  }
}

8. Podsumowanie

• Callable nie jest Runnable - nie należy o nim myśleć jako o mającym związek z Runnable, jest raczej jego zamiennikiem - w tym sensie, że to takie "Runnable co ma wynik",
• zadanie (task) może być Callable lub Runnable (ale w przypadku Callable to jest kompozycja!),
• Future  (w szczególności FutureTask, które jest Runnable!) możemy: odpytywać o stan wykonania, zabijać lub też pobierać wynik,
• wykonawcom można podać Callable, ale i Runnable (i w szczególności FutureTask, co daje nam w każdym przypadku kolekcję w pełni zarządzanych zadań!),
• jeszcze raz: FutureTask jest Runnable, a w związku z tym może być argumentem metody submit Execut-orów,
• FutureTask po wykonaniu (lub przerwaniu) woła metodę done(). Dobre miejsce na handlery wyników!
• ThreadPool ExecutorService po otrzymaniu zadań do wykonania przez submit (nie ważne czy Callable czy Runnable) zwraca  FutureTask. Bez specjalnego tworzenia (obiektów typu Future) może tu być i cancel i odpytywanie o koniec zadania!

9. Zadania