虎の穴 J2SE 5.0 入門 Concurrency Utilities No.1

Concurrency Utilities その一非同期処理の巻

マルチスレッドは得意ですか ?

マルチスレッドのアプリケーションはいろいろ時をつけなくてはいけないことがあります。同期や同時アクセスやプライオリティ、デッドロックなどなど。

Java では安易にスレッドを作ることができるのでいつのまにかスレッドをたくさん使っていたなんてことも起こりがちですが、そのつけは大きいです。

そんなマルチスレッドのアプリケーションを作る際に福音書となっているのが Java スレッドプログラミングです。残念ながら今は品切れで入手が困難なので、もし本屋に残ってたらすぐにでも購入したほうがいいです。

マルチスレッドと同様この本も内容が難しくて、なかなか理解するのが大変なんですが、マスターできたら鬼に金棒状態です。そして、この本の作者が Concurrency Utilities の原型を作った Doug Lea なのです。

Concurrency Utilities では次のような機能を提供しています。

非同期実行
スレッドプール
時間表現
キュー、ブロッキングキュー
ロック
同期処理セマフォなど
アトミック変数

その他に Thread クラスの変更なども含まれています。

結構なボリュームがあるので、とりあえずここでは非同期実行、スレッドプールを取り上げたいと思っています。残りの項目に関してはその 2 以降で。

非同期処理とは

普段、プログラミングをしていても、同期だとか非同期とかはあまり気にすることはありません。というのも、ほとんど同期処理しか行わないからです。

ようするにメソッドをコールしたら、そのメソッドの処理が終わるまで待たなくてはいけないということです。

逆にいえば、非同期処理は珍しいのです。その珍しい例をあげてみると

SwingUtilities#invokeLater
ServerSocketChannel#accept, SocketChannel#connect, read, write

ぐらいしか思いつきません ^^;;

SwingUtilities#invokeLater メソッドは普段最も使われる非同期処理ではないでしょうか。たとえば、ボタンが押されたときにあまり時間がかかる処理の場合、そこでウンスンになってしまうので後で行わせるというものです。でも、実際にはこれは純粋な非同期処理ではありません。というのも、結局処理するスレッドは同じだからです。

次のようなプログラムだとボタンが押されたら、ボタンが押された状態のままになってしまいます。これは結局 Swing のイベントスレッドが invokeLater メソッドで指定された処理も行ってしまうからです (本当の理由はもうちょっと複雑ですが...)。

import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;
import javax.swing.JButton;
import javax.swing.JFrame;
import javax.swing.SwingUtilities;
 
import static javax.swing.JFrame.*;
 
public class HeavyButton {
    public HeavyButton() {
        JFrame frame = new JFrame("Heavy Button...");
        frame.setDefaultCloseOperation(EXIT_ON_CLOSE);
 
        JButton button = new JButton("Heavy");
        button.addActionListener(new ActionListener() {
           public void actionPerformed(ActionEvent event) {
               SwingUtilities.invokeLater(new Runnable() {
                   public void run() {
                       try {
                           Thread.sleep(10000L);
                       } catch (InterruptedException ex) {
                           ex.printStackTrace();
                       }
                   }
               });
            }
        });
		
        frame.getContentPane().add(button);
        frame.pack();
        frame.setVisible(true);
    }

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

非同期処理を書かなくてはいけないとしたらどう書くか考えて見ましょう。

普通は Runnable インタフェースを使って、別のスレッドで処理を行わせます。

    Runnable runnable = new Runnable() {
        public void run() {
            // ...   非同期にしたい何らかの処理
        }
    };
	
    Thread thread = new Thread(runnable);
    thread.start();

別スレッドで行わせている処理が終わったかどうかを調べるためにはイベントを使ったり、wait ... notify を使ったりします。たとえば、wait ... notify を使うとすると次のようになります。

public class AsyncSample {
    public AsyncSample() {}
 
    public void foo() {
        Runnable runnable = new Runnable() {
            public void run() {
                // ...   非同期にしたい何らかの処理
                synchronized (AsyncSample.this) {
                    AsyncSample.this.notifyAll();
                }
            }
        };
	
        Thread thread = new Thread(runnable);
        thread.start();
  
        synchronized (this) {
            try {
                wait();
            } catch (InterruptedException ex) {}
        }
 
        System.out.println("非同期処理終了");
    }
 
    public static void main(String[] args) {
        new AsyncSample().foo();
    }
}

wait ... notify を使うと synchronized にしたり、いろいろ面倒くさそうです。

その面倒くさいところをやってくれるのが Concurrency Utilities なのです。

Executor

非同期処理を行うためにキーとなるのが java.util.concurrent.Executor インタフェースです。後から Executor を派生させた ExecutorService というインタフェースも出てきますが、それは後の話。

Executor インタフェースには次のメソッドが定義されています。単にこれだけです。

public void execute(Runnable command);

Executor インタフェースの JavaDoc にはこのインタフェースをインプリメントしたクラスのサンプルが出ています。一番簡単なのが次に示す DirectExecutor クラスです。

import java.util.concurrent.Executor;
 
public class DirectExecutor implements Executor {
    public void execute(Runnable r) {
        r.run();
    }
}

Runnable インタフェースの run メソッドをコールしているだけです。Thread#start メソッドではなくて、Runnable#run メソッドです。

これを使って簡単なサンプルを作ってみました。

サンプルのソース

ExecutorTest1.java
DirectExecutor.java
ConcurrentUtils.java

import java.util.concurrent.Executor;
 
public class ExecutorTest1 {
    public ExecutorTest1() {
        System.out.println("Initial Threads:");
        ConcurrentUtils.listThread();
 
        Executor executor = new DirectExecutor();
        System.out.println("Processing Threads:");
        ConcurrentUtils.listThread();
 
        executor.execute(new RunnableTask());
        System.out.println("RuunableTask has Done.");
 
        System.out.println("Terminating Threads:");
        ConcurrentUtils.listThread();
    }
 
    class RunnableTask implements Runnable {
        public void run() {
            try {
                Thread.sleep(3000L);
            } catch (InterruptedException ex) {
                System.out.println("RunnableTask is Canceled");
                return;
            }
            System.out.println("RunnableTask is Done.");
        }
    }
 
    public static void main(String[] args) {
        new ExecutorTest1();
    }
}

何回か出てくる ConcurrentUtils#listThread メソッドはスレッドの一覧を表示するために作ったメソッドで次のような実装になっています。

    public static void listThread() {
        ThreadGroup group = Thread.currentThread().getThreadGroup();
        while (true) {
            ThreadGroup parent = group.getParent();
            if (parent == null) {
                break;
            } else {
                group = parent;
            }
        }
 
        group.list();
    }

ThreadGroup クラスには、自分が管理しているスレッドグループおよびスレッドの一覧を表示する list メソッドがあります。これを利用しています。

さて、ExecutorTest1 を実行してみましょう。

C:\JSR166\examples>java ExecutorTest1
Initial Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
Processing Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
RunnableTask is Done.
RuunableTask has Done.
Terminating Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
 
C:\JSR166\examples>

あれっと思いませんか。というのもスレッド数はプログラムを通して全然変わっていないからです。

このプログラムは実をいうと非同期でもなんでもありません。executor メソッドを呼び出しても、単に run メソッドを呼ぶだけなので同期で実行します。

これじゃｙおもしろくないですね。

どうやら、マルチスレッドで非同期処理を行うには、Executor が自分でスレッドを用意しなくてはいけないようです。

そこで、やはり Executor インタフェースの JavaDoc にのっている ThreadPerTaskExecutor クラスを使ってみました。

import java.util.concurrent.Executor;
 
public class ThreadPerTaskExecutor implements Executor {
    public void execute(Runnable r) {
        new Thread(r, "ThreadPerTaskExecutor").start();
    }
}

execute メソッドがコールされると、毎回スレッドを作って処理を実行させます。

この Executor を使って先ほどの ExecutorTest1 クラスを書きかえたのが、ExecutorTest1_1 クラスです。

サンプルのソース

ExecutorTest1_1.java
ThreadPerTaskExecutor.java

これを実行してみましょう。

C:\JSR166\examples>java ExecutorTest1_1
Initial Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
Processing Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[ThreadPerTaskExecutor,5,main]
RunnableTask is Done.
Terminating Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
 
C:\JSR166\examples>

これだと RunnableTask#run メソッドは ThreadPerTaskExecutor スレッドで行われていることが分かります。

でも、毎回スレッドを作るのも嫌な感じです。次章ではそこいらへんを考えてみましょう。

スレッドプールを使ってみよう

タスクがあるごとにスレッドを作っていたら、毎回生成するのも時間がかかりますし、スレッドの管理をするにもやりにくくなって、いいことはあまりありません。

毎回スレッドを作るのがよくないならば、使いまわせばいいですね。この方式をスレッドプールといいます。

スレッドを常に用意しておいて、必要なときにプールしておいたスレッドを使用するわけです。使い終わったスレッドはまた次のタスクが登録されるまで遊ばせておきます。

今までスレッドプールを作るのはそれなりに面倒くさかったのですが、Concurrency Utilities にはスレッドプールをすぐに使えるようなメソッドが用意されています。

Executors クラスで定義されている

newSingleThreadExecutor
newFixedThreadPool
newCachedThreadPool
newScheduledThreadPool

がスレッドプールを生成するためのメソッドになります。この他にも newSingleThreadScheduledExecutor という長い名前のスレッドプールを作成するメソッドもあります。

いずれのメソッドも static なので、いつでもどこでもスレッドプールを生成することができます。

これらの中で一番簡単な newSingleThreadExecutor メソッドを使用して、先ほどの ExecutorTest1 クラスを書きかえてみましょう。

サンプルのソース

ExecutorTest2.java

先ほどと違うのは、Executor オブジェクトを生成する部分だけです。

        Executor e = Executors.newSingleThreadExecutor();

これで、自分で Executor インタフェースの実装クラスを書く必要はありません。

さて、実行してみましょう。タスクを実行しているスレッドが pool-1-thread-1 と表示されているのがお分かりでしょうか。

C:\JSR166\examples>java ExecutorTest2
Initial Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
Processing Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]
RunnableTask is Done.
Terminating Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]

うまくいったと思ったのもつかの間、このサンプルは終わらないのです。

どうやら、新たなタスクが登録されるのを待っている状態にあるらしいです。しかし、それを制御するためのメソッドは Executor インタフェースにはありません。困った。

こういうときはやっぱり Javadoc。もういちど、newSingleThreadExecutor メソッドを見てみました。

そうしたら、newSingleThreadExecutor メソッドの戻り値は Executor オブジェクトではないことに気がつきました。ExecutorService インタフェースになっています。ExecutorService インタフェースはスレッドプールなどに対応した Executor インタフェースの派生インタフェースとなっているようです。

そこで、ExecutorService インタフェースの Javadoc を見てみます。そうしたら、ありました。shutdown メソッドと shutdownNow メソッドです。この両者ともシャットダウンの処理は同じようですが、登録だけされてまだ実行していないタスクを戻り値として返すかどうかの違いのようです。

このメソッドを main メソッドの最後に入れたものが ExecutorTest2_1 クラスです。

サンプルのソース

ExecutorTest2_1.java

実行してみると、ちゃんと終了するのが確認できると思います。

C:\JSR166\examples>java ExecutorTest2_1
Initial Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
RuunableTask Threads;
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]
RunnableTask is Done.
Processed Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]
Final Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
 
C:\JSR166\examples>

shutdown メソッドをコールしてからスレッドの一覧を表示しているのですが、pool-1-thread-1 スレッドがなくなっているのが確認できます。

さて、ExecutorService オブジェクトを生成させたメソッド newSingleThreadExecutor メソッドは名前のとおり、1 つのスレッドでタスクを処理するはずです。それを確かめるために複数のタスクを処理させてます。

サンプルのソース

ExecutorTest2_2.java

単に RunnableTask オブジェクトを複数生成して、execute メソッドをコールしているだけです。

        ExecutorService e = Executors.newSingleThreadExecutor();
         
        e.execute(new RunnableTask());
 
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }
 
        System.out.println("RuunableTask1 Threads;");
        ConcurrentUtils.listThread();
 
        e.execute(new RunnableTask());
 
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }
 
        System.out.println("RuunableTask2 Threads;");
        ConcurrentUtils.listThread();

これを実行してみます。関係のある部分だけ下に示しました。

RunnableTask starts.
RuunableTask1 Threads;
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]
RunnableTask is Done.
RunnableTask starts.
RuunableTask2 Threads;
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]
RunnableTask is Done.

シングルスレッドなので、タスクが同時に処理されることはないはずです。上の例でも execute メソッドは、前の処理が終わる前にコールしているのですが、前のタスクの終了後に処理されていることが分かります。

また、どちらのタスクも pool-1-thread-1 スレッドで実行されていることも分かります。

Runnable インタフェースと Callable インタフェース

newSingleThreadExecutor メソッド以外のメソッドで生成されるスレッドプールはちょっと置いておいて、先に ExecutorService インタフェースをもう少し詳しく見ていきましょう。

そうすると、execute メソッド以外に submit メソッドというのがあることが分かります。しかし、処理内容は execute メソッドと何が違うのでしょう。

答えは戻り値を戻すかどうかということにあります。sumbit メソッドの戻り値は Future オブジェクトです。

java.util.concurrent.Future インタフェースの Javadoc を見てみると、登録されたタスクがどのような状態にあるかなどを調べることができるインタフェースのようです。

isDone メソッドでタスクが終了しているかどうか、isCancelled メソッドがタスクがキャンセルされたかを示すようです。また、cancel メソッドでタスクをキャンセルすることもできるようです。

これらのメソッドの意味は分かるのですが、分からないのが get メソッドです。結果を取得できるというのですが... 何の結果なのでしょう?

まぁ、難しいことは考えずに、使ってみましょう。

サンプルのソース

ExecutorTest3.java

このサンプルでは isDone メソッドでタスクが終了しているかどうかを調べてみました。

また、何の結果だがよく分かりませんが、とりあえず get メソッドをコールしてみます。

        ExecutorService e = Executors.newSingleThreadExecutor();
 
        Future future = e.submit(new RunnableTask());
        System.out.println("RuunableTask is Called");
        System.out.println("Future#isDone(): " + future.isDone());
 
        try {
            System.out.println("Future#get(): " + future.get());
            System.out.println("Future#isDone(): " + future.isDone());
        } catch (ExecutionException ex) {
            ex.printStackTrace();
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }

実行してみると、まず

RuunableTask is Called
Future#isDone(): false

と表示されます。そして、しばらくすると

RunnableTask is Done.
Future#get(): null
Future#isDone(): true

と表示されます。

get メソッドをコールする前にスリープなのは入れていないのですが、get メソッドでブロックしているようです。

Javadoc を読んでみると、get メソッドはタスクが処理するまでブロックすると書いてあります。今ままでのサンプルはタスクがいつ終わるか分からなかったので、適当にスリープを入れておいたのですが、get メソッドを入れておけばタスクの終了を知ることが出来るようです。

また、タスクが終了したら isDone メソッドの戻り値が true になるので、これでタスクが終了したかどうかも調べることが可能になります。

しかし、get メソッドの戻り値は null。結局、何の結果だかよく分からないままです ^^;;

分からないことは先延ばしして、Future#cancel メソッドを使ってみます。

サンプルのソース

ExecutorTest3_1.java

submit メソッドの戻り値の Future オブジェクトに対して cancel メソッドをコールしているだけです。

        ExecutorService e = Executors.newSingleThreadExecutor();
 
        Future future = e.submit(new RunnableTask());
        System.out.println("RuunableTask is Called");
        System.out.println("Future#isCancelled(): " + future.isCancelled());
        System.out.println("Future#isDone(): " + future.isDone());
 
        future.cancel(true);
        System.out.println("Future#isCancelled(): " + future.isCancelled());
        System.out.println("Future#isDone(): " + future.isDone());

実行して見ます。

RuunableTask is Called
Future#isCancelled(): false
Future#isDone(): false
RunnableTask is Canceled
java.lang.InterruptedException: sleep interruptedFuture#isCancelled(): true
        at java.lang.Thread.sleep(Native Method)
        at ExecutorTest3_1$RunnableTask.run(ExecutorTest3_1.java:25)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:41
7)
        at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:269)
        at java.util.concurrent.FutureTask.run(FutureTask.java:123)
        at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExec
utor.java:650)
Future#isDone(): true

        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor
.java:675)
        at java.lang.Thread.run(Thread.java:595)

スタックトレースと普通の出力が交じってしまって見にくいですが、InterruptedException が発生していることはよく分かると思います。すなわち、cancel メソッドがスレッドの interrupt を呼び出しているわけです。

ということは、InterruptedException を正しく処理していない場合や、InterruptedException が発生しないような処理では cancel メソッドをコールしても処理は止まらない可能性もあるということです。

Javadoc の cancel メソッドの説明にも This attempt will fail if the task has ... could not be cancelled for some other reason という記述があり、必ずしも cancel できるとは限らないようです。

さて、先延ばしにしていた get メソッドの戻り値についてもう一度調べてみます。やはり、困ったときはやはり Javadoc に戻りましょう。

ExecutorService インタフェースの submit メソッドを見てみると、引数に Runnable オブジェクトと任意の型 (Generics を利用しています) の result というメソッドがオーバロードされています。

result という引数の名前が怪しいですね。

そこで、ExecutorTest3 を、このオーバロードされた submit メソッドに変更してみました。

サンプルのソース

ExecutorTest3_2.java

result には単純に true を入れています。

        ExecutorService e = Executors.newSingleThreadExecutor();
 
        Future<Boolean> future = e.submit(new RunnableTask(), true);
        System.out.println("RuunableTask is Called");
        System.out.println("Future#isDone(): " + future.isDone());
 
        try {
            System.out.println("Future#get(): " + future.get());
            System.out.println("Future#isDone(): " + future.isDone());
        } catch (ExecutionException ex) {
            ex.printStackTrace();
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }

実行したらどうなると思いますか。

RuunableTask is Called
Future#isDone(): false
RunnableTask is Done.
Future#get(): true
Future#isDone(): true

おぉ、ちゃんと true が戻ってきました。これで、タスク処理が終わったかどうかを判定させることができます。

けど、本当にそんな使い方のために submit メソッドがあるのでしょうか...

ちがいます ^^;;

submit メソッドにはもう 1 つオーバロードがあります。それは Callable オブジェクトを引数にとるものです。

この Callable インタフェースというのは何なんでしょう?

2003 年の JavaOne で Doug Lea はこの 2 つのインタフェースの違いを次のように書いています。

Runnable : Actions

Callable : Functions

よく分かりません ^^;; それでも、Callable インタフェースのソースを見ると違いがわかります。Callable インタフェースは次のように定義されています。

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

メソッド名が run ではなくて call ということよりも大きい違いがあります。

それは Callable#call メソッドには戻り値があるということです。

Runnable#run メソッドには戻り値がないので、非同期処理の結果を受け取るにはそれなりの工夫が必要です。それに対して Callable#call メソッドは戻り値があるので非同期処理の結果を戻すのが簡単だという違いがあるのです。

ということは、Future#get メソッドの戻り値の結果というのは call メソッドの戻り値と仮定することができます。

さっそく、サンプルを作ってこれを確かめてみましょう。

サンプルのソース

ExecutorTest4.java

call メソッドは、現在時刻を返すようにしてみました。

    public ExecutorTest4() {
        ExecutorService e = Executors.newSingleThreadExecutor();

        Future future = e.submit(new CallableTask());
        System.out.println("CallableTask is Called");
        System.out.println("Future#isDone(): " + future.isDone());

        try {
            System.out.println("Future#get(): " + future.get());
            System.out.println("Future#isDone(): " + future.isDone());
        } catch (ExecutionException ex) {
            ex.printStackTrace();
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }

        e.shutdown();
    }

    class CallableTask implements Callable<Date> {
        public Date call() {
            try {
                Thread.sleep(1000L);
            } catch (InterruptedException ex) {
                System.out.println("CallableTask is Canceled");
                return null;
            }

            System.out.println("CallableTask is Done.");
            return new Date();
        }
    }

実行するとどうなるでしょうか。

CallableTask is Called
Future#isDone(): false
CallableTask is Done.
Future#get(): Mon Sep 06 23:26:51 JST 2004
Future#isDone(): true

どうやら仮説はあっていたようです。ちゃんと call メソッドの戻り値を get メソッドで取得することができました。

今まで、非同期処理は Runnable インタフェースしかなかったので、処理の結果を戻すためにはクラスフィールドにするなどの工夫が必要でした。

しかい、Callable インタフェースを使えば、そういう面倒くさいことを考えずに、非同期処理の結果を扱うことができるようになります。

この違いが Doug Lea のいうところの Actions と Functions なんでしょうね。

タスクの大量投入

Runnable インタフェースと Callable インタフェースの違いが分かったので、ExecutorService インタフェースに話を戻しましょう。

ExecutorService インタフェースの Javadoc を見ていると invokeAll と invokeAny という何やら怪しげなメソッドがあります。これらのメソッドを使ってみましょう。

サンプルのソース

ExecutorTest5.java

invokeAll メソッドは引数が Callable インタフェースのコレクションになっているので、登録したタスクをすべて実行するのでしょう。

        ExecutorService e = Executors.newSingleThreadExecutor();
 
        List<Callable<String>> tasks = new ArrayList<Callable<String>>();
        for (int i = 0; i < 5; i++) {
            Callable<String> task = new CallableTask("task" + i);
            tasks.add(task);
        }
 
        try {
            List<Future<String>> futures = e.invokeAll(tasks);
 
            for (Future<String> future: futures) {
                try {
                    System.out.println("Future#get(): " + future.get());
                } catch (ExecutionException ex) {
                    ex.printStackTrace();
                }
            }
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }

CallableTask クラスは次のように単にスリープして自分の名前を返すようにしています。

    class CallableTask implements Callable {
        private String name;
 
        public CallableTask(String name) {
            this.name = name;
        }
 
        public String call() {
            System.out.println(name + " is Called.");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException ex) {}
 
            System.out.println(name + " exits.");
            return name;
        }
    }

Executors#newSingleThreadExecutor メソッドでスレッドプールを取得しているので、シーケンシャルに実行されることが予想されますが、どうでしょう。

C:\JSR166\examples>java ExecutorTest5
task0 is Called.
task0 exits.
task1 is Called.
task1 exits.
task2 is Called.
task2 exits.
task3 is Called.
task3 exits.
task4 is Called.
task4 exits.
Future#get(): task0
Future#get(): task1
Future#get(): task2
Future#get(): task3
Future#get(): task4

C:\JSR166\examples>

invokeAll メソッドですべてのタスクが実行されることが分かります。しかもシーケンシャルに実行されていることも分かるのですが、すべての処理が終わらないと Future#get メソッドから戻らないというのもおもしろいですね。そんなもんなのでしょうか。

次は、もう 1 つの invokeAny メソッドです。

サンプルのソース

ExecutorTest5_1.java

invokeAny の Any というのはどういう意味なのでしょうか。All だったらすべてですが... しかも、invokeAny メソッドの戻り値はコレクションではなく Callable#call の戻り値になっています。

            String result = e.invokeAny(tasks);
            System.out.println("Result: " + result);

CallableTask クラスは先ほどと同じにしています。

C:\JSR166\examples>java ExecutorTest5_1
task0 is Called.
task0 exits.
task1 is Called.
Result: task0
task1 exits.

C:\JSR166\examples>

Any なので、1 しか実行されないのかと思ったのですが、どうやら違ったようです。とりあえず処理できるだけ処理させておいて、処理結果が戻ってきたらその時点で終わりにしているようです。

これはどういうときに使うのでしょうね? よく分かりません。

普通のスレッドプール

今までは Executors#newSingleThreadExecutor メソッドを使用して ExecutorService オブジェクトを取得していたのでシングルスレッドでシーケンシャルにタスクを処理していました。でも、これじゃあまりスレッドプールっぽくないですね。

その他の newFixedThreadPool メソッドや newCachedThreadPool メソッドを使って、普通のスレッドプールを体験してみます。

スレッドプールの実現には 2 種類の方法が考えられます。

ある決められた数のスレッドを作成しておき、それ以上スレッドは増加させない。スレッドの数以上のタスクが登録された場合は、待ち状態にする。
タスクが登録されたときに待ち状態のスレッドがあればそのスレッドでタスクを処理する。待ち状態にあるスレッドがない場合は新たにスレッドを生成してタスクを処理させる。

1 の方法は Executors#newFixedThreadPool メソッド、2 の方法は Executors#newCachedThreadPool メソッドが対応しています。

サンプルのソース

ExecutorTest6.java

はじめは Executors#newFixedThreadPool メソッドを使用してみます。

        ExecutorService e = Executors.newFixedThreadPool(3);
 
        System.out.println("Initial Thread Pool:");
        ConcurrentUtils.listThread();
         
        e.execute(new RunnableTask("Task 0"));
 
        System.out.println("Thread Pool:");
        ConcurrentUtils.listThread();
 
        e.execute(new RunnableTask("Task 1"));
 
        try {
            Thread.sleep(3000L);
        } catch (InterruptedException ex) {
            ex.printStackTrace();
        }
 
        System.out.println("Processed Threads:");
        ConcurrentUtils.listThread();
 
        e.shutdown();
 
        System.out.println("Terminating Threads:");
        ConcurrentUtils.listThread();

Fixed なのでスレッドの上限が決まってはずです。その時々のスレッドを見ていればそれが確認できます。

C:\JSR166\examples>java ExecutorTest6
Initial Thread Pool:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
Thread Pool:
java.lang.ThreadGroup[name=system,maxpri=10]
   Task 0 Starts.
 Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]
Task 1 Starts.
Task 0 is Done.
Task 1 is Done.
Processed Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]
        Thread[pool-1-thread-2,5,main]
Terminating Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
 
C:\JSR166\examples>

はじめはタスク処理用のスレッドがなかったのが、タスクを登録するたびに増えていく様子が分かるでしょうか。

Task 0 は pool-1-thread-1 で処理され、Task 1 は pool-2-thread-2 で処理されています。

タスクがスレッドの上限を超えてしまったらどうなるでしょうか。

サンプルのソース

ExecutorTest6_1.java

次のように 5 つのタスクを登録してみました。上限は 3 なのでどうなるでしょうか。

        for (int i = 0; i < 5; i++) {
            e.execute(new RunnableTask("Task" + i));
        }

結果は予想できますね。

C:\JSR166\examples>java ExecutorTest6_1
Initial Thread Pool:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
Task0 Starts.
Task1 Starts.
Task2 Starts.
Task0 is Done.
Task3 Starts.
Task1 is Done.
Task4 Starts.
Task2 is Done.
Task3 is Done.
Task4 is Done.
Processed Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]
        Thread[pool-1-thread-2,5,main]
        Thread[pool-1-thread-3,5,main]
Terminating Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
 
C:\JSR166\examples>

予想通りスレッドは 3 しか作られませんでした。一度に処理のできないタスクは待ち状態にあり、実行中のタスクが終了すると、次の待ち状態にあるタスクの処理がはじまっています。

次は Executors#newCachedThreadPool メソッドでスレッドプールを作ってみましょう。

サンプルのソース

ExecutorTest7.java

以下のコードのようにどんどんタスクを投入してみます。

        ExecutorService e = Executors.newCachedThreadPool();
 
        System.out.println("Initial Thread Pool:");
        ConcurrentUtils.listThread();
         
        for (int i = 0; i < 5; i++) {
            e.execute(new RunnableTask("Task" + i));
        }

たぶん、 5 つタスクを登録すれば、5 つのスレッドができると思うのですが...

C:\JSR166\examples>java ExecutorTest7
Initial Thread Pool:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
Processed Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
Task0 Starts.
Task1 Starts.
Task2 Starts.
Task3 Starts.
Task4 Starts.
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]
        Thread[pool-1-thread-2,5,main]
        Thread[pool-1-thread-3,5,main]
        Thread[pool-1-thread-4,5,main]
        Thread[pool-1-thread-5,5,main]
Task0 is Done.
Task1 is Done.
Task2 is Done.
Task3 is Done.
Task4 is Done.
Terminating Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]

C:\JSR166\examples>

予想通り 5 つのスレッドがパラレルに走って、タスクを処理しています。

しかし、このようにタスクを登録していったらどんどんスレッドが増えていってしまうのでしょうか。

サンプルのソース

ExecutorTest7_1.java

ほぼ同時にタスクを登録するのがいけないのであって、すこし間隔をあけてタスクを登録してみます。

        for (int i = 0; i < 5; i++) {
            e.execute(new RunnableTask("Task" + i));
            try {
                Thread.sleep(1500L);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
        }

こうすれば、最後の方にタスクを登録するときには、はじめの方のタスクは終了しているはずなので、スレッドが使いまわされるはずです。

C:\JSR166\examples>java ExecutorTest7_1
Initial Thread Pool:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
Task0 Starts.
Task1 Starts.
Task0 is Done.
Task2 Starts.
Task1 is Done.
Task3 Starts.
Task2 is Done.
Task4 Starts.
Task3 is Done.
Task4 is Done.
Processed Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]
        Thread[pool-1-thread-2,5,main]
Terminating Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]

C:\JSR166\examples>

予想通り、2 つのスレッドだけでタスク処理が行われました。

ところで、これらのスレッドは使われなくなったらどうなるのでしょうか。実をいうと newFixedThreadPool メソッドも newCachedThreadPool メソッドも同じクラスのオブジェクトを返すだけです。

Executors クラスの該当部分を抜き出してみます。

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue());
    }
 
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue());
    }

この java.util.concurrent.ThreadPoolExecutor クラスのコンストラクタは

第 1 引数	:	初期状態におけるスレッド数
第 2 引数	:	スレッド数の最大値
第 3 引数	:	スレッドがタスク処理をしない状態で生き残る時間
第 4 引数	:	第 3 引数の単位
第 5 引数	:	タスクを登録するためのキュー

なので、スレッドが生き抜くのは第 3 引数で指定される時間になります。

newFixedThreadPool メソッドの場合ははじめから指定されたスレッドを作成し、それが増減することはありません。

newCachedThreadPool メソッドでは、初期状態では 0 個のスレッド、最大は int で表される最大値になります。また、60 秒間タスクがなければスレッドは消滅します。

これを試してみましょう。

サンプルのソース

ExecutorTest7_2.java

タスクを登録してから 20 秒ごとにスレッドを表示してみます。

        e.execute(new RunnableTask("Task0"));
 
        for (int i = 0; i <= 80; i+=20) {
            System.out.println("\n" + i + " seconds:");
            ConcurrentUtils.listThread();
            try {
                TimeUnit.SECONDS.sleep(20);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
        }

少し長いのですが実行結果を示します。

C:\JSR166\examples>java ExecutorTest7_2
Initial Thread Pool:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]

0 seconds:
java.lang.ThreadGroup[name=system,maxpri=10]
Task0 Starts.
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]
Task0 is Done.

20 seconds:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]

40 seconds:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]

60 seconds:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
        Thread[pool-1-thread-1,5,main]

80 seconds:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]
Terminating Threads:
java.lang.ThreadGroup[name=system,maxpri=10]
    Thread[Reference Handler,10,system]
    Thread[Finalizer,8,system]
    Thread[Signal Dispatcher,10,system]
    java.lang.ThreadGroup[name=main,maxpri=10]
        Thread[main,5,main]

C:\JSR166\examples>

60 秒の時はまだワーキングスレッドが残っていますが、80 秒の時はなくなっています。

キープアライブ時間は 60 秒なのに 60 秒の時にスレッドが残っているのは変な感じを受けるかもしれません。しかし、よく考えてみるとタスクがなくなってから 60 秒なのです。タスク処理には 2 秒必要なので、60 秒の時にはまだ残っているわけですね。

まだあるスレッドプール

Executors クラスの Javadoc を見ているとまだスレッドプールを生成するためのメソッドがあることが分かります。

Executors#newSingleThreadScheduledExecutor メソッドとは Executors#newScheduledThreadPool は戻り値として ScheduledExecutorService オブジェクトを返します。

名前からして java.util.Timer クラスのような遅延処理や繰り返し処理を行うだろうということは予想できます。Timer クラスだとワーキングスレッドは 1 つなのですが ScheduledExecutorService インタフェースはスレッドプールに対応しているので複数のスレッドで繰り返し処理を行うことができそうです。

まずは遅延処理からです。

サンプルのソース

ExecutorTest8.java

タスクを遅延処理するためには execute メソッドではなく schedule メソッドを使用します。schdule メソッドの第 1 引数は Runnable オブジェクトもしくは Callable オブジェクト、第 2 引数が遅延させる時間、第 3 引数が時間の単位です。

        ScheduledExecutorService e = Executors.newScheduledThreadPool(1);
 
        e.schedule(new RunnableTask("Task"), 2000, TimeUnit.MILLISECONDS);
        System.out.println("Task is called at " + new Date());

この例だと 2 秒後にタスクが処理されます。

ここで使用しているタスクも示しておきます。

        public void run() {
            System.out.println(name + " starts at " + new Date());
 
            try {
                Thread.sleep(1000L);
            } catch (InterruptedException ex) {
                System.out.println(name + " is Canceled");
                return;
            }
 
            System.out.println(name + " is done at " + new Date());
        }

単にコールされたときの時間を表示し、抜けるときにも時間を表示するというものです。

C:\JSR166\examples>java ExecutorTest8
Task is called at Wed Sep 15 23:01:17 JST 2004
Task starts at Wed Sep 15 23:01:19 JST 2004
Task is done at Wed Sep 15 23:01:20 JST 2004
		
C:\JSR166\examples>

実行すると schedule メソッドがコールされたのが 23:01:17 なのに対して、実際に RunnableTask#run がコールされたのは 23:01:19 とちゃんと 2 秒後になっているのが分かります。

遅延はそれほど難しくはありません。

よく分からないのが繰り返し処理です。何がよく分からないかというと英語です ^^;;

繰り返し処理を行う場合、scheduleAtFixedRate メソッドと scheduleWithFixedDelay という 2 つのメソッドを使用します。

それにしても at XXX rate と with XXX delay の違いは何なのでしょう? 全然分かりません。こういうときに英語のネイティブな方はいいですよね。すぐ違いが分かるのですから。

まぁ、愚痴をいってもしょうがないので、とりあえずサンプル作って何が違うか見てみましょう。

サンプルのソース

ExecutorTest9.java

1 秒間スリープするタスクを 2 秒周期で scheduleAtFixedRate メソッドと scheduleWithFixedDelay メソッドの両方を使用して登録してみました。

        ScheduledExecutorService e = Executors.newScheduledThreadPool(10);
 
        ScheduledFuture future = e.scheduleAtFixedRate(
                                      new RunnableTask("Task AT Fixed Rate"), 
                                      2000, 2000, TimeUnit.MILLISECONDS);
        System.out.println("Task is called at " + new Date());

        try {
            Thread.sleep(10000L);
        } catch (InterruptedException ex) {}

        future.cancel(true);

        try {
            Thread.sleep(2000L);
        } catch (InterruptedException ex) {}

        future = e.scheduleWithFixedDelay(new RunnableTask("Task WITH Fixed Delay"), 
                                          2000, 2000, TimeUnit.MILLISECONDS);
        System.out.println("Task is called at " + new Date());

まずは scheduleAtFixedRate メソッドを使ったほうの実行結果です。

C:\JSR166\examples>java ExecutorTest9
Task is called at Wed Sep 15 23:15:02 JST 2004
Task AT Fixed Rate starts  at Wed Sep 15 23:15:04 JST 2004
Task AT Fixed Rate is done at Wed Sep 15 23:15:05 JST 2004
Task AT Fixed Rate starts  at Wed Sep 15 23:15:06 JST 2004
Task AT Fixed Rate is done at Wed Sep 15 23:15:07 JST 2004
Task AT Fixed Rate starts  at Wed Sep 15 23:15:08 JST 2004
Task AT Fixed Rate is done at Wed Sep 15 23:15:09 JST 2004
Task AT Fixed Rate starts  at Wed Sep 15 23:15:10 JST 2004
Task AT Fixed Rate is done at Wed Sep 15 23:15:11 JST 2004
Task AT Fixed Rate starts  at Wed Sep 15 23:15:12 JST 2004

タスクは 2 秒後とに処理されていることが分かります。

次に scheduleWithFixedDelay メソッドを使ったほうです。


Task is called at Wed Sep 15 23:17:06 JST 2004
Task WITH Fixed Delay starts  at Wed Sep 15 23:17:08 JST 2004
Task WITH Fixed Delay is done at Wed Sep 15 23:17:09 JST 2004
Task WITH Fixed Delay starts  at Wed Sep 15 23:17:11 JST 2004
Task WITH Fixed Delay is done at Wed Sep 15 23:17:12 JST 2004
Task WITH Fixed Delay starts  at Wed Sep 15 23:17:14 JST 2004
Task WITH Fixed Delay is done at Wed Sep 15 23:17:15 JST 2004

C:\JSR166\examples>

よく見てください。違いが分かりますか。

scheduleWithFixedDelay メソッドを使った場合、3 秒おきにタスクが処理されているのがお分かりでしょうか。

どうやら、この 2 つのメソッドの違いはタスク処理が開始する間隔とタスク処理間の間隔ということにありそうです。

このため scheduleAtFixedRate メソッドを使用すれば、タスク処理時間にかかわらず定期的に処理を行うことができます。しかし、タスク処理時間 > period の場合、処理は遅延してしまうようです。

逆にscheduleWithFixedDelay メソッドの場合、タスク処理にいくら時間がかかってもタスクとタスクの間の時間は変化しません。

このような性質の差があることがわかれば、使い分けもできそうです。

おわりに

Executor インタフェースと Executors クラスを使えば、面倒くさい非同期処理やスレッドプールも思いのままです。いままでちょっと手を出しにくかった方も、これならば全然 OK ですね。

しかし、ちょっと不安も残ります。それは RuntimeException の問題です。非同期処理中に RuntimeException が発生すると、だれも catch できないのでスレッドが落ちてしまうという問題です。J2SE 1.4 までは ThreadGroup#uncaughtException メソッドを使用してだれも catch しなかった RuntimeException を捕らえることができました。

しかし、ThreadPoolExecutor クラスで生成されるスレッドはメインスレッドと同じスレッドグループなので、uncaughtException メソッドをオーバライドすることはできません。

どうすればいいか。答えはそのニへ。

今回使用したサンプルはここからダウンロードできます。

ConcurrencyUtils1.zip

参考

JCP JSR-166 http://jcp.org/en/jsr/detail?id=166
仕様書はここからダウンロードできます
Doug Lea の Web サイト
http://gee.cs.oswego.edu/dl/concurrency-interest/
JSR-166 の原型もあります。また、JSR-166 に関する ML などもあります。

(Feb. 2004)
(全面改訂 Sep. 2004)

Concurrency Utilities その一 非同期処理の巻

Concurrency Utilities その一非同期処理の巻