Groovy で Thread-Per-Message パターン

『増補改訂版 Java言語で学ぶデザインパターン入門 マルチスレッド編』に載っているデザインパターンを Groovy/GPars で書こうシリーズ、今回は Thread-Per-Message パターン。　このパターンは、メッセージ（リクエスト）が送られてきたら、それに対して1つスレッドを割り当てて処理しよう、ってパターンです。　メッセージ毎にスレッドを割り当てて並行処理するため、1度に複数のメッセージが送られてきてもそれなりに対応できます（限度はあるけど）。　まぁ、完全に Web サーバーのイメージですね。

サンプルでは Host クラスに対してメッセージを送り（request() メソッドを呼び出す）、それに対する処理を別スレッドで行っています。　1つ目のサンプルでは、各メッセージに対して単純にスレッドを作成・開始します。　2つ目のサンプルでは java.util.concurrent パッケージの Executor, ExecutorService インターフェースを使用して、柔軟なスレッド割り当てを行えるようにしています。　3つ目のサンプルでは GPars のライブラリを使って同様のことを行っています。

サンプル・コード　その1

まずは基本となるサンプル。　作成するソースは

• Host クラス　リクエストをスレッドに割り振る
• Helper クラス　リクエストに対する処理を行う
• 実行スクリプト

Host クラス

Host クラスは request() メソッドが呼ばれると処理を行う、簡単なサーバーです。　リクエスト処理のロジックはこのクラスではなく、後の Helper クラスに定義し、Host クラスは単にそのクラスのメソッド呼び出すだけです：

class Host{
    
    private final Helper helper = new Helper()

    void request(int count, String s){
        println "    request($count, $s) BEGIN"
        Thread.start { helper.handle(count, s) }    // 処理は Helper クラスに
        println "    request($count, $s) END"
    }
}

Host クラスはリクエストをどのように並行処理するかを決めます。　この実装では単に各リクエストに対してスレッドを作成して開始しているだけです。

Helper クラス

Helper クラスは実際に処理を実装するクラス。　上記の Host クラスの実装だとこのクラスが状態を持つとマズイですが、まぁ必要なら各スレッドごとにインスタンスを生成してやれば問題ないかと。　ここではこんな感じの処理：

class Helper {

    void handle(int count, String s){
        println "        handle($count, $s) BEGIN"
        count.times{ Thread.sleep(100); print s }
        println()
        println "        handle($count, $s) END"
    }
}

実行スクリプト

実行スクリプトはスレッドのことなんて気にせずに、単にオブジェクト生成とメソッド呼び出しをするだけです：

println 'main BEGIN'

def host = new Host()
host.request(10, 'A')
host.request(20, 'B')
host.request(30, 'C')

println 'main END'

実行スクリプトで逐次的にメソッド呼び出しをしても、処理は並行に行われて、各処理の完了や結果のレスポンス（このサンプルではレスポンスがないですが）の順序はマチマチになります。

サンプル・コード　その2 ： Executor(ExecutorService) を使うサンプル

次は java.util.concurrent パッケージのクラスを用いて書き換えたサンプル。　ソースは

• Helper クラスは同じ
• Host クラスの代わりに ExecutorHost クラス
• 実行スクリプト

ExecutorHost クラス

ExecutorHost クラスは Executor オブジェクトを保持し、各リクエストに対してこの Executor オブジェクトを介して処理のスレッドへの割り振りや実行を行います。　そもそもスレッドを作成して実行しているかどうかも気にする必要がありません。

import java.util.concurrent.Executor

class ExecutorHost {

    private final Helper helper = new Helper()
    private final Executor executor

    public ExecutorHost(Executor executor){
        this.executor = executor;
    }

    public void request(int count, String s){
        println "    request($count, $s) BEGIN"
        this.executor.execute { helper.handle(count, s) }    // Groovy のクロージャは Runnable を実装
        println "    request($count, $s) END"
    }
}

Executor#execute() メソッドは Runnable オブジェクトを引数にとりますが、Groovy のクロージャは Runnable を実装しているのでそのまま引数として渡しています。

実行スクリプト

次は実行スクリプト。　ExecutorHost クラスはコンストラクタに Executor オブジェクトを渡す必要があるので、Executros クラス（Executor のユーティリティ・クラス）を使って Executor オブジェクトを作成しています。

import java.util.concurrent.Executors

println 'main BEGIN'

def exeService = Executors.newCachedThreadPool()
def host = new ExecutorHost(exeService)
try{
    host.request(10, 'A')
    host.request(20, 'B')
    host.request(30, 'C')

}finally{
    exeService.shutdown()
    println 'main END'
}

ここでは Executor インターフェースを拡張した ExecutorService インターフェースのインスタンスを生成しているので、リクエストが終わったら ExecutorService#shutdown() メソッドを呼び出す必要があるので注意。　例外が投げられた場合も呼び出されなければならないので、finally 節に書いてます。

サンプル・コード　その3 ： GParsExecutorsPool を使うサンプル

最後はもう少し Groovy らしく、GPars を使ったサンプル。　但し、変更するのは実行スクリプトのみで、GPars でサポートされている ExecutorService オブジェクトを使うように変更しただけのものです。

import groovyx.gpars.GParsExecutorsPool
import java.util.concurrent.ExecutorService

GParsExecutorsPool.withPool { ExecutorService service ->
    println "main BEGIN"

    def host = new ExecutorHost(service)
    host.request(10, 'A')
    host.request(20, 'B')
    host.request(30, 'C')

    println "main END"
}

まぁ、有難味は ExecutorService#shutdown() メソッドをこちらで呼ぶ必要がないことでしょうか。　GPars では GParsExecutorsPool の他に GParsPool というのもあるようですが、こちらで使用されるスレッドプールは Executor, ExecutorService を実装していないのでここでは使っていません。　性能的には GParsPool の方が良いようです。

増補改訂版 Java言語で学ぶデザインパターン入門 マルチスレッド編

プログラミングGROOVY