高価なオーバヘッドをともなうプロセス全体の状態 （仮想メモリマップ、ファイル記述子など）をセーブするかわりに、 オーバヘッドの少ないコンテキストスイッチ （レジスタやスタックポインタ、プログラムカウンタのみをセーブする） が同じアドレス空間内でおこなわれます。
スレッドとプロセスはよく似ていますが、プロセスと異なり、 複数のスレッドが同じアドレス空間を共用できます。 スレッドは、メソッドの局所変数以外の変数とメソッドを共用できます。 すなわち、プロセスと比べるとはるかに軽量なので、 １つのアプリケーションで数百のスレッドを同時に実行することも可能です。
単一プロセッサのハードウェアでも、 スレッドを使った方がプログラムの実行速度が速い場合があります。 それは、I/O が発生している間にも実行できる計算処理がある場合などです。
スレッドを使う３つの理由は以下のとおりです。
↓ 実行文１ ┌---------┴---------┐
↓
| ||||||||||
新しいスレッドを生成する方法は２つあります。
java.lang.Thread の run メソッドは、以下のとおり何も実装されていません。
従って run() をオーバーライドする必要があります。
public class Thread implements Runnable {
public void run() {
if (target != null) {
target.run();
}
}
}
class mango extends Thread {
public void run() { ....}
}
class mango implements Runnable {
public void run() { ....}
}
Runnable インタフェースは単に次のようなものです。
public interface Runnable {
public abstract void run();
}
Runnable インタフェースを実装したクラスは、 Thread を拡張したクラスとほとんど同じように機能します。 違いは、Runnable インタフェースには run メソッドしかないのに対して、 Thread クラスには run() 以外にも stop(),sleep(),suspend(), resume(),setPriority(),getPriority() などのメソッドがあることです。
| java Core API の java.lang.Thread と java.lang.Runnable を見てみましょう。 |
run() は、任意のコードを本体に持てる必須メソッドです。 run() メソッドは、public で、引数を取らず、戻り値はなく、 例外を投げることは禁止されています。
新しいスレッドは、生成された時点では実行を開始しません。
スレッドは、start メソッドを呼ぶことで実行を開始し、 run メソッドから実行を開始します。 run() は実行を開始する場所で、start() がそれを実行させます。 すなわち run() は、 新しく作成されたスレッドにとっての main() と同じです。
class example {
public static void main(String[] a) {
// alternative 1
ExtnOfThread t1 = new ExtnOfThread();
t1.start();
// alternative 2
Thread t2 = new Thread (new ImplOfRunnable());
t2.start();
}
}
class ExtnOfThread extends Thread {
public void run() {
System.out.println("Extension of Thread running ");
try { sleep(10);}
catch (InterruptedException ie) {return;}
}
}
class ImplOfRunnable implements Runnable {
public void run() {
System.out.println("Implementation of Runnable running ");
Thread t = Thread.currentThread();
try { t.sleep(10);}
catch (InterruptedException ie) {return;}
}
}
| 説明 |
上記プログラムについて簡単に説明します。
t1 は Thread を拡張した ExtnOfThread のオブジェクトです。 |
例外と同様、Thread のサブクラスを作るとき名前を指定できます。 これを実現するには、文字列を受け取って親クラスのコンストラクタに渡す、 コンストラクタを書かなければなりません。
class NewThread extends Thread {
NewThread(String s) {super(s);} // constructor
public void run(){....}
}
スレッドは、以下に列挙した条件のいずれかが成立すると、終了します。
スレッドを stop() メソッドで殺すこともできます。 stop() メソッドはスレッドの実行を止め、スレッドオブジェクトを破壊します。 再実行するには、新しいスレッドを再度作らなければなりません。 一時的にスレッドを停止し、再開するには suspend() と resume() を使います。
制御を他のスレッドに頻繁に明渡すためには yield メソッドを使います。 yield は、自身の実行を一時中断して、他のスレッドに実行権を渡します。
複数のスレッド間の関係は同期 (synchronization) と呼ばれます。
複数のスレッドを用いるプログラムは、
スレッド間で必要とされる同期の種類によって、以下の４つに分類されます。
class PingPong extends Thread {
String word;
int delay;
PingPong(String what,int delayTime) {
word = what;
delay = delayTime;
}
public void run() {
try {
for (;;) {
System.out.print(word + " ");
sleep(delay);
}
} catch (InterruptedException e) {
return ;
}
}
public static void main(String[] args) {
new PingPong("ping", 33).start();
new PingPong("pong",100).start();
}
}
testPrimeは、 問題を分割して同じデータ構造の別の部分を、 複数スレッドを使って処理するものです。スレッドは互いに干渉しません。
// demonstrates the use of threads to test a number for primality
class testPrime {
static final int MAX_ARGS = 1; // test number
public static void main(String[] s) {
if (s.length != MAX_ARGS) usage(s.length);
long possPrime = Long.parseLong(s[0]);
int centuries = (int)(possPrime/100) + 1;
for (int i=0; i< centuries; i++) {
new testRange(i*100,possPrime).start();
}
}
static void usage(int argno) {
System.out.println("Usage : java testPrime NUMBER");
System.exit(argno);
}
}
class testRange extends Thread {
static long possPrime;
long from,to;
testRange(int argFrom,long argposPrime) {
possPrime = argposPrime;
if (argFrom == 0) from = 2;
else from = argFrom;
to = argFrom + 99;
}
public void run() {
for (long i=from; i<=to && i< possPrime; i++) {
if (possPrime%i == 0) {
System.out.println("factor "+i+" found by thread "+
this.getName());
this.stop();
}
yield();
}
}
}
osaka-32] java testPrime 2048 factor 2 found by thread Thread-2 factor 512 found by thread Thread-7 factor 1024 found by thread Thread-12 factor 128 found by thread Thread-3 factor 256 found by thread Thread-4
| 質問 | このプログラムはなにをしているでしょうか？ |
public class Press {
static int pressureGauge=0;
static final int SAFETY_LIMIT=20;
static final int CHECK_LIMIT =15;
static final int MAX_PRE =10;
public static void main(String[] a) {
pressure [] p1 = new pressure[MAX_PRE];
for (int i=0; i< MAX_PRE; i++) {
p1[i] = new pressure();
p1[i].start();
}
try {
for (int i=0; i< MAX_PRE; i++) {
p1[i].join();
}
} catch (Exception e) {}
System.out.println("Gauge reads "+pressureGauge+",safe limit is "
+SAFETY_LIMIT);
}
}
class pressure extends Thread {
void RaisePresure() {
if (Press.pressureGauge < Press.SAFETY_LIMIT - Press.CHECK_LIMIT) {
try {
sleep(100);
} catch (Exception e) {}
Press.pressureGauge += Press.CHECK_LIMIT;
} else;
}
public void run() {
RaisePresure();
}
}
osaka-49] java Press Gauge reads 150,safe limit is 20
| 質問 | 圧力を増やす前に常にチェックしているのに、 なぜ大幅に安全限界を越えてしまったのでしょうか？ |
これは、データの競合または競合状態と呼ばれる例です。
競合状態は２つ以上のスレッドが同じ値を同時に更新すると発生します。 競合状態を防ぐには、 複数のスレッドが同じデータにアクセスする時には相互排除しなければなりません。
java では、スレッドの相互排他制御はオブジェクトに作りこまれています。
synchronized キーワードを用いると、
いついかなるときでも、
多くても１つのスレッドだけが、
そのオブジェクトのロックを取得しているよう
java の実行システムが保証しています。
synchronized キーワードは、クラス、メソッド、 コードブロックのそれぞれにつけられます。 いずれの場合も、指定されたオブジェクトの排他ロックが取得され、 それからコードが実行され、その後ロックが解放されます。 ロックが他のスレッドによってすでに取得されていた場合には、 ロックを取得しようとするスレッドは、 そのロックが解放されるまでサスペンド（待ち）します。
synchronized なコード領域は必要最小限にとどめるべきです。
それは、排他制御が性能上の制約になる場合があるからです。
では、synchronized をクラス、メソッド、コードブロックにつけてみましょう。
static なクラスメソッドに synchronized を付けることで実現できます。
このクラスメソッドは、同時にはただ１つしか動けないので同期できます。
static synchronized void RaisePresure() {}
PressUp は、 オブジェクトがいくつ作られても、クラス全体で１つのメソッドしか動けないので、 圧力計にアクセスし、 更新するコードをシリアライズ(逐次化）できます。
class PressUp {
static int pressureGauge=0;
static final int SAFETY_LIMIT= 20;
static final int CHECK_LIMIT =15;
static final int MAX_PRE =10;
public static void main(String[]args) {
PressureUp []p1 = new PressureUp[MAX_PRE];
for (int i=0; i< MAX_PRE; i++) {
p1[i] = new PressureUp();
p1[i].start();
}
try{ for (int i=0;i< MAX_PRE;i++) p1[i].join(); } catch(Exception e){}
System.out.println("Gauge reads "+pressureGauge+", safe limit is "
+SAFETY_LIMIT);
}
}
class PressureUp extends Thread {
static synchronized void RaisePressure() {
if (PressUp.pressureGauge < PressUp.SAFETY_LIMIT - PressUp.CHECK_LIMIT) {
// wait briefly to simulate some calculations
try{sleep(100);} catch (Exception e){}
PressUp.pressureGauge += PressUp.CHECK_LIMIT;
} else ; // pressure too high -- don't add to it.
}
public void run() {
RaisePressure();
}
}
osaka-53] java PressUp Gauge reads 15, safe limit is 20
コードブロックの前に synchronized を付けることで実現できます。
その領域に入る前にロックしなければならないオブジェクトを、
括弧でくくって明示的に書かなければなりません。
void RaisePressure() {
static Object O = new Object();
synchronized(0) {
if (PressUp.pressureGauge < PressUp.SAFETY_LIMIT - PressUp.CHECK_LIMIT){
// wait briefly to simulate some calculations
try{sleep(100);} catch (Exception e){}
PressUp.pressureGauge += PressUp.CHECK_LIMIT;
} else ; // pressure too high -- don't add to it.
}
}
static でないメソッドに synchronized を付けることで実現できます。
この場合は、相互排他に使うロックを提供するオブジェクトは明示されていませんが、 this オブジェクトです。
synchronized void foo() {...}
は、以下と等価です。
void foo() {
synchronized (this) {....}
}
従って、すべてのスレッドが自分のロックを取得できるので、 スレッド間の排他制御には使えません。 同一のオブジェクトに同時に呼ばれるかもしれないメソッドがある場合には便利です。
同じデータを処理する複数のスレッドがあるので、同期させなければなりません。 それだけでなく、 データが準備できていないときは一時中断するような機構が必要です。
synchronized で同期させると同時に、一時中断と再開させるメソッド wait/notify/notifyAllを使います。 これらのメソッドは、synchronized コードの中でしか使えません。 つまり排他ロックを保持している時しか呼べないことを意味します。
これらは、Object クラスに定義されており、すべてのクラスが継承します。
| wait() | 処理するデータがないので、できるまでロックを解除して処理を中断します。 |
| notify() | データができたので、ロックを解除して自分は処理を中断し、 中断中のスレッドのいずれかを再開させます。 |
| notifyAll() | データができたので、ロックを解除して自分は処理を中断し、 中断中のスレッドすべてを再開させます。 |
ここで古典的な２つのスレッド Producer と Consumer について紹介します。
import java.util.*;
class Producer extends Thread {
static final int MAX_QUEUE = 5;
static final int SLEEP_PRO = 1000;
private Vector messages = new Vector();
public void run() {
try {
while (true) {
putMessage();
sleep(SLEEP_PRO);
}
}
catch (InterruptedException e) {}
}
private synchronized void putMessage() throws InterruptedException {
while (messages.size() == MAX_QUEUE) wait();
messages.addElement( new Date().toString());
notifyAll();
}
public synchronized String getMessage() throws InterruptedException {
while (messages.size() == 0) wait();
String msg = (String) messages.firstElement();
messages.removeElement(msg);
notifyAll();
return msg;
}
}
class Consumer extends Thread {
static final int SLEEP_CON = 2000;
Producer producer;
Consumer(Producer p) {
producer = p;
}
public void run() {
try {
while (true) {
String msg = producer.getMessage();
System.out.println(" Got message : " + msg);
sleep(SLEEP_CON);
}
} catch (InterruptedException e) {}
}
}
class Model {
public static void main(String[] args) {
Producer producer = new Producer();
producer.start();
new Consumer(producer).start();
}
}
osaka-30] java Model Got message : Thu Oct 05 11:17:42 JST 2000 Got message : Thu Oct 05 11:17:43 JST 2000 Got message : Thu Oct 05 11:17:44 JST 2000 Got message : Thu Oct 05 11:17:45 JST 2000 Got message : Thu Oct 05 11:17:46 JST 2000 Got message : Thu Oct 05 11:17:47 JST 2000 Got message : Thu Oct 05 11:17:48 JST 2000 Got message : Thu Oct 05 11:17:49 JST 2000 Got message : Thu Oct 05 11:17:50 JST 2000 Got message : Thu Oct 05 11:17:51 JST 2000 Got message : Thu Oct 05 11:17:52 JST 2000 Got message : Thu Oct 05 11:17:54 JST 2000 Got message : Thu Oct 05 11:17:56 JST 2000 Got message : Thu Oct 05 11:17:58 JST 2000
| 説明 |
上記プログラムについて簡単に説明します。
ふつう Producer は作ったデータを境界つきのバッファにいれます。 ということは、Producer がバッファを満たしてしまうと、 バッファに空きがでるまで wait() する必要があります。 また Consumer は、バッファ上のものを取り除いたら、 Producer に notify()/notifyAll() で知らせる必要があります。 Producer は１秒間隔でデータを作りますが、 Consumer は２秒間隔でデータを消費します。 しかし Producer はキューがいっぱいになると wait しなければなりません。 従って、だんだんデータをつくる間隔が２秒になっていきます。 あるスレッドから、他の sleep しているスレッドを java.lang.Thread.interrupt() メソッドを呼んで割り込むことができます。 こうすると、割り込まれたスレッドは起きますが、 notify()/notifyAll() で起きた場合と区別できる必要があります。 このため割り込まれた場合は、そのスレッドで例外 InterruptedException を検出できます。 また例外が起きた時はそれがわかるように、 エラーメッセージなどを出すようにしましょう。 |
次に、Consumer を２つにしてみましょう。
import java.util.*;
import Producer;
class ConsumerUp extends Thread {
static final int SLEEP_CON = 2000;
Producer producer;
String name;
ConsumerUp(String name,Producer p) {
this.producer = p;
this.name = name;
}
public void run() {
try {
while (true) {
String msg = producer.getMessage();
System.out.println(name + " got message : " + msg);
sleep(SLEEP_CON);
}
} catch (InterruptedException e) {}
}
}
class ModelUp {
public static void main(String[] args) {
Producer producer = new Producer();
producer.start();
new ConsumerUp("One",producer).start();
new ConsumerUp("Two",producer).start();
}
}
osaka-29] java ModelUp One got message : Thu Oct 05 11:16:58 JST 2000 Two got message : Thu Oct 05 11:16:59 JST 2000 One got message : Thu Oct 05 11:17:00 JST 2000 Two got message : Thu Oct 05 11:17:01 JST 2000 One got message : Thu Oct 05 11:17:02 JST 2000 Two got message : Thu Oct 05 11:17:03 JST 2000 One got message : Thu Oct 05 11:17:04 JST 2000 Two got message : Thu Oct 05 11:17:05 JST 2000 One got message : Thu Oct 05 11:17:06 JST 2000 Two got message : Thu Oct 05 11:17:07 JST 2000 One got message : Thu Oct 05 11:17:08 JST 2000 Two got message : Thu Oct 05 11:17:09 JST 2000 One got message : Thu Oct 05 11:17:10 JST 2000 Two got message : Thu Oct 05 11:17:11 JST 2000 One got message : Thu Oct 05 11:17:12 JST 2000 Two got message : Thu Oct 05 11:17:13 JST 2000
synchronized は、完全な解決策ではありません。 システムがデッドロック に陥ってしまうことがありうるからです。 デッドロックとは、複数のスレッドの間で、資源を保有しているものと、 要求しているものとの間に循環した依存関係のある状態のことです。 例えば、スレッド A が X をロックしていて Y をロックしようとしている時、 スレッド B が Y をロックしていて X のロックが解放されるのを待っている状態です。
この場合、システムはその部分で停止します。 これは、１つの synchronized なメソッドが別の synchronized なメソッドを呼ぶと簡単に発生します。
デッドロックはユーザの責任で回避しなければなりません。
| まとめ |
| メソッド名 | 概 要 | クラス名 |
|---|---|---|
| run() | スレッドの実行コードを定義します。 | Thread |
| start() |
このスレッドに実行を開始させる。 Java 仮想マシンがこのスレッドのrun メソッドを呼び出す。 | Thread |
| stop() | スレッドの実行を強制的に停止させる。 | Thread |
| wait() | 処理するデータがないので、できるまでロックを解除して処理を中断します。 | Thread |
| yield() | 現在実行中のスレッドオブジェクトを一時的に休止させ、 他のスレッドの実行を許可する。 | Thread |
| join() | スレッドが死ぬのを待機する。 | Thread |
| resume() | 一時停止されたスレッドを再開する。 | Thread |
| suspend() | このスレッドを中断する。 | Thread |
| notify() | オブジェクトのモニタで待機中の 1 つのスレッドを再開する。 | Object |
| notifyAll() |
オブジェクトのモニタで待機中のすべてのスレッドを再開する。 中断中のスレッドすべてを再開させます。 | Object |