Worker Threadパターン

workerというのは「作業者」という意味。WorkerThreadパターンでは、ワーカースレッドが仕事を一つずつ取りに行き、処理を行う。仕事がなかったら、ワーカスレッドは新しい仕事が届くまで待つ。Worker Threadのことを、Background Threadと呼ぶこともある。

次のような例を考える。ClientThreadクラスのスレッドが、Channelクラスに仕事のリクエストを出す。Channelクラスのインスタンスはワーカースレッドを5人かかえている。ワーカースレッドは、仕事のリクエストがくるのを待っている。仕事のリクエストがやってくると、ワーカスレッドは、Channelから仕事のリクエストを1個もらって処理する。処理が終わるとワーカースレッドはChannelのところに戻ってきて次のリクエストを待つ。

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

Main.java

public class Main { 
    public static void main(String[] args) { 
        Channel channel = new Channel(5);   // ワーカースレッドを5個作成
        channel.startWorkers(); // ワーカースレッドをスタートさせる
        new ClientThread("Alice", channel).start();  // 仕事のリクエストを出すスレッドをスタートさせる
        new ClientThread("Bobby", channel).start(); 
        new ClientThread("Chris", channel).start(); 
    } 
}

ClientThread.java

public class ClientThread extends Thread { 
    ...
    public void run() { 
        try { 
            for (int i = 0; true; i++) { 
                Request request = new Request(getName(), i); 
                channel.putRequest(request);  // 仕事のリクエストを出す
                Thread.sleep(random.nextInt(1000)); 
            } 
        } catch (InterruptedException e) { 
        } 
    } 
}

Channel.java

public class Channel { 
    private static final int MAX_REQUEST = 100; 
    private final Request[] requestQueue; 
    private int tail;
    private int head;
    private int count;

    private final WorkerThread[] threadPool; 

    public Channel(int threads) { 
        this.requestQueue = new Request[MAX_REQUEST]; 
        this.head = 0; 
        this.tail = 0; 
        this.count = 0; 

        threadPool = new WorkerThread[threads]; 
        for (int i = 0; i < threadPool.length; i++) { 
            threadPool[i] = new WorkerThread("Worker-" + i, this); 
        } 
    }

    public void startWorkers() { 
        for (int i = 0; i < threadPool.length; i++) { 
            threadPool[i].start(); 
        } 
    }

    public synchronized void putRequest(Request request) { 
        while (count >= requestQueue.length) { 
            try { 
                wait(); 
            } catch (InterruptedException e) { 
            } 
        } 
        requestQueue[tail] = request; 
        tail = (tail + 1) % requestQueue.length; 
        count++; 
        notifyAll(); 
    }

    public synchronized Request takeRequest() { 
        while (count <= 0) { 
            try { 
                wait(); 
            } catch (InterruptedException e) { 
            } 
        } 
        Request request = requestQueue[head]; 
        head = (head + 1) % requestQueue.length; 
        count--; 
        notifyAll(); 
        return request; 
    } 
}

WorkerThread.java

public class WorkerThread extends Thread { 
    ...
    public void run() { 
        while (true) { 
            Request request = channel.takeRequest(); // 仕事を取りに行く
            request.execute(); 
        } 
    } 
}

Channelクラスは仕事のリクエストの受け渡しのためにrequestQueueというフィールドを持っている。このキューにリクエストを入れるのはputRequestメソッド、取り出すのはtakeRequestメソッド。ここでは、Producer-Consumerパターンが使われている。Thread-Per-Messageパターンでは、仕事を実行するたびに新しいスレッドを起動していた。しかしWorker Threadパターンでは、ワーカースレッドが繰り返し仕事を実行するので、新しいスレッドを起動する必要がない。WorkerThreadが持っているフィールドは、自分が仕事のリクエストを得るためのChannelのインスタンスだけで、具体的な仕事内容は何も知らない。

登場人物

Client役
Client役は、仕事のリクエストをRequest役として作成し、Channel役に渡す。サンプルプログラムでは、ClientThreadクラスがこの役をつとめた。

Channel役
Channel役は、Client役からRequest役を受け取り、Worker役に渡す。サンプルプログラムでは、Channelクラスがこの役をつとめた。

Worker役
Worker役はChannel役からRequest役をもらい、その仕事を実行する。仕事が終わったら、次のRequest役をもらいにいく。サンプルプログラムでは、WorkerThreadクラスがこの役をつとめた。

Request役
Request役は、仕事を表すためのもの。Request役は、その仕事を実行するのに必要な情報を保持している。サンプルプログラムでは、Requestクラスがこの役をつとめた。

考えを広げるためのヒント

スループットの向上
Thread-Per-Messageパターンと違ってWorkerThreadパターンではスレッドを使いまわし、リサイクルしてスループットを上げることがテーマのひとつ。

キャパシティの制御
同時に提供できるサービスの量をキャパシティと呼ぶ。Worker役の数は自由に定めることができる。サンプルプログラムでは、Channelのコンストラクタに与える引数threadsがそれにあたる。Worker役の数が多ければ、それだけ並行して仕事をすることができるが、同時期にリクエストされる仕事の数よりも多いWorker役を用意しても役に立たず消費するリソースも増えてしまう。Worker役の数は、必ずしも起動時に定めておく必要はなく、動的に変化させることもできる。Channel役はRequest役を保持している。Channel役が保持しておくことのできるRequest役の数が多いと、Client役とWorker役の処理速度のずれを埋める(緩衝する)ことができる。しかし、Request役を保持するためには大量のメモリを消費することになる。ここには、キャパシティとリソースのトレードオフがあることがわかる。

invocationとexecutionの分離
通常のメソッド呼び出しを行うとき、「メソッドの起動」と「メソッドの実行」は続けて行われる。通常のメソッド呼び出しでは、起動と実行は不可分である。しかし、Worker ThreadパターンやThread-Per-Messageパターンは、メソッドの起動と実行を分離している。メソッドの起動はinvocation、実行はexecutionという。Worker ThreadパターンやThread-Per-Messageパターンは、メソッドのinvocationとexecutionを分離しているといってもよい。これにより以下のメリットが生まれる。

応答性の向上
executionに時間がかかても、invocationをした方は先に進んでいくことができる。
実行順序の制御(スケジューリング)
Request役に優先順位を設け、Channgel役がWorker役にRequest役を渡す順番を制御できる。これを要求のスケジューリングという。
キャンセル可能・繰り返し実行可能
invocationしたがexecuionはキャンセルという機能が実現できる。
分散処理
invokeするコンピュータとexecueするコンピュータを分けやすくなる。

多態的なRequest役
たとえば、Requestクラスのサブクラスを作り、そのインスタンスをChannelに渡したとしても、WorkerThreadは問題なくそのインスタンスのexecuteをよんでくれる。これは多態性(ポリモルフィズム)を利用していると表現できる。仕事を実行するのに必要な情報は、Request役にすべて記述されているため、多態的なRequest役を作って仕事の種類を増やしてもChannel役やWorker役側は修正する必要がない。仕事の種類が増えても、Worker役は相変わらずRequest役のexecuteメソッドを呼ぶだけである。

『Java言語で学ぶデザインパターン(マルチスレッド編)』まとめ(その１０)

Worker Threadパターン

登場人物

考えを広げるためのヒント