Groovy で Guarded Suspension パターン

『増補改訂版 Java言語で学ぶデザインパターン入門マルチスレッド編』に載っているデザインパターンを Groovy/GPars で書こうシリーズ、今回は Guarded Suspension パターン。　Guarded Wait, Spin Lock などとも呼ばれるそうです。

このパターンは java.util.concurrent パッケージに導入された BlockingQueue インターフェースとその実装クラスを使えばあまり何も考える必要はありません。　まぁ、それだけで話が終わると素っ気ないので、BlockingQueue インターフェースのサブタイプにどんなものがあるか見ていきましょう。　ちなみに BlockingQueue に定義されているメソッドは「Java SE 6 のコレクション・フレームワークのメソッド (5) : BlockingQueue, BlockingDeque」などを参照。

BlockingQueue のサブタイプ

BlockingQueue のサブインターフェースには TransferQueue*1, BlockingQueue があります。　で、それぞれの実装クラスは以下のようなものがあります：

BlockingQueue
- ArrayBlockingQueue
- DelayQueue
- LinkedBlockingQueue
- PriorityBlockingQueue
- SynchronousQueue
TransferQueue
- LinkedTransferQueue
BlockingDeque
- LinkedBlockingDeque

BlockingQueue のサブクラスにはいろいろあって、それぞれに特性があるようですが、普通に BlockingQueue のインスタンスが必要なら LinkedBlockingQueue クラスを使えばいいんじゃないでしょうか。

サンプル・コード

作成するソースは

リクエストを表す Requesut
リクエストを作成する ClientThread
リクエストを処理する ServerThread
実行スクリプト

ClientThread, ServerThread はそれぞれ独自スレッドで動作します。　また、リクエストをストックしておくキューとして BlockingQueue をそのまま用います。

Request クラス

必須ではありませんが、ここでは Request クラスを Immutable として作成します：

@groovy.transform.Immutable
class Request {
    String name
}

ClientThread クラス

ClientThread は20個の Request オブジェクトを作ってキューに入れます。　普通に作成するのもアレなので、1個作成するたびにランダムな期間（800ミリ秒以下）だけスリープします：

import java.util.concurrent.BlockingQueue

class ClientThread extends Thread{
    
    private final BlockingQueue<Request> requestQueue
    private final Random random
    
    ClientThread(BlockingQueue<Request> queue, String name, long seed){
        super(name)
        this.requestQueue = queue
        this.random = new Random(seed)
    }

    @Override
    void run() {
        (0..<20).each{ int i ->
            def request = new Request("No. $i")
            println "${Thread.currentThread().name} requests $request"
            
            try{
                this.requestQueue.put(request)
                Thread.sleep this.random.nextInt(800)

            }catch(InterruptedException ex){
                ex.printStackTrace()
            }
        }
    }
}

ServerThread クラス

ServerThread クラスは ClientThread クラスとだいたい同じですが、Request オブジェクト取り出すところが違います：

import java.util.concurrent.BlockingQueue

class ServerThread extends Thread{

    private final BlockingQueue<Request> requestQueue
    private final Random random

    ServerThread(BlockingQueue<Request> queue, String name, long seed){
        super(name)
        this.requestQueue = queue
        this.random = new Random(seed)
    }

    @Override
    void run() {
        20.times{
            try{
                def request = this.requestQueue.take()
                println "${Thread.currentThread().name} handles $request"
                Thread.sleep this.random.nextInt(1000)

            }catch(InterruptedException ex){
                ex.printStackTrace()
            }
        }
    }
}

実行スクリプト

実行スクリプトではキューとして LinkedBlockingQueue オブジェクトを生成し、ClientThread, ServerThread オブジェクトも作成して、これらはスレッドを start() させます：

import java.util.concurrent.LinkedBlockingQueue

def requestQueue = new LinkedBlockingQueue<Request>()
new ClientThread(requestQueue, 'Alice', 3141592L).start()
new ServerThread(requestQueue, 'Bobby', 6535897L).start()