Producer-Consumerパターン

producer(生産者)はデータを作成するスレッド、consumer(消費者)はデータを利用するスレッドのこと。生産者と消費者が別のスレッドとして動くとき、両者の処理速度のズレが問題になる。消費者がデータを受け取ろうとしたときにデータがまだ作り出されていなかったり、生産者がデータを渡そうとしたときに消費者がデータを受け取れる状態ではなかったりする。Producer-Consumerパターンでは、生産者と消費者の間に「橋渡し役」が入る。この橋渡し役がスレッド間の処理速度のズレを埋める。生産者と消費者の両方が単数の場合、Pipeパターンと呼ぶことがある。

3人のコックがケーキを作りテーブルに置く、それを3人のお客が食べる、テーブルには3個までケーキが置ける、という例を考える。テーブルの上に3個のケーキが置かれているのに、コックがテーブルにさらにケーキを置こうとしたなら、コックはテーブルに置く場所ができるまで待たされる。テーブルの上に1個もケーキが置かれていない時に、お客がテーブルからケーキを取ろうとしたなら、お客はケーキが置かれるまで待たされる。

(コード全体は本書を参照のこと)

MakerThread

public class MakerThread extends Thread { 
    ...
    private static int id = 0; // ƒケーキ通し番号(全コックで共通)
    ...
    public void run() { 
        try { 
            while (true) { 
                Thread.sleep(random.nextInt(1000)); 
                String cake = "[ Cake No." + nextId() + " by " + getName() + " ]"; 
                table.put(cake); 
            } 
        } catch (InterruptedException e) { 
        } 
    } 
    private static synchronized int nextId() { 
        return id++; 
    } 
}

EaterThread


public class EaterThread extends Thread { 
    ...
    public void run() { 
        try { 
            while (true) { 
                String cake = table.take(); 
                Thread.sleep(random.nextInt(1000)); 
            } 
        } catch (InterruptedException e) { 
        } 
    }
    ...

Table

public class Table { 
    private final String[] buffer; 
    private int tail;  // 次にputする場所
    private int head;  // 次にtakeする場所
    private int count; // buffer内のケーキ数
    ...
    // ケーキを置く
    public synchronized void put(String cake) throws InterruptedException { 
        System.out.println(Thread.currentThread().getName() + " puts " + cake); 
        while (count >= buffer.length) { 
            wait(); 
        } 
        buffer[tail] = cake; 
        tail = (tail + 1) % buffer.length; 
        count++; 
        notifyAll(); 
    } 
    // ケーキを取る
    public synchronized String take() throws InterruptedException { 
        while (count <= 0) { 
            wait(); 
        } 
        String cake = buffer[head]; 
        head = (head + 1) % buffer.length; 
        count--; 
        notifyAll(); 
        System.out.println(Thread.currentThread().getName() + " takes " + cake); 
        return cake; 
    } 
}

実行結果

...
MakerThread-1 puts [ Cake No.10 by MakerThread-1 ]
EaterThread-1 takes [ Cake No.10 by MakerThread-1 ]
MakerThread-1 puts [ Cake No.11 by MakerThread-1 ]
EaterThread-3 takes [ Cake No.11 by MakerThread-1 ]
MakerThread-3 puts [ Cake No.12 by MakerThread-3 ]
MakerThread-3 puts [ Cake No.13 by MakerThread-3 ]
EaterThread-2 takes [ Cake No.12 by MakerThread-3 ]
EaterThread-3 takes [ Cake No.13 by MakerThread-3 ]
...

登場人物

Data役：
Producer役によって作成され、Consumer役によって利用される。上の例では、ケーキがこれにあたる。
Producer役：
Data役を作成してChannel役に渡す。上の例では、MakerThreadクラスがこれにあたる。
Consumer役：
Channel役からData役を受け取って利用する。上の例では、EaterThreadクラスがこれにあたる。
Channel役：
Producer役からData役を受け取り、保管、Consumer役からの求めに応じてData役を渡す。Producer役とConsumer役からのアクセスに対して排他制御を行う。Channel役は、Producer役とConsummer役の間に入って、Data役を渡す中継地点の役割、通信路の役割を果たす。上の例では、Tableクラスがこれにあたる。

考えを広げるヒント

マルチスレッドの動作への配慮っを行っているコードは、すべてChannel役であるTableクラスの中に隠されている。
Producer役はConsumer役の処理の進み具合に左右されない。
InterruptedExceptionという例外がthrowされる可能性がある、ということは、「時間がかかる」メソッドである、「キャンセルできる」メソッドであると言える。

スレッドがwaitで待っている場合にも、sleepと同じでキャンセルできる。interruptメソッドを使うと、waitしているスレッドに対して、「もうnotify/notifyAllを待たなくていいよ。ウェイトセットから出ておいで」と伝えることができる。しかし、waitの場合には、ロックに注意する必要がある。スレッドはウェイトセットにはいるときに、一旦ロックを解放していた。wait中にinterruptを呼ばれたスレッド(つまりキャンセルされたスレッド)は、ロックを再び取り直してからInterruptedExceptionを投げる。つまり、ロックが取れるまでは例外InterruptedExceptionを投げることはできない。