Java Advent Calendar 2014 リソースごとに同時アクセス数の制限を行う

これはJava Advent Calendar 2014の6日目の記事です。

昨日はbiblichorさんの「Jersey2×SpringSecurityのハマりポイント三選」でした。

Webシステム等、多数の処理を同時に行う場合、処理に使用するリソース毎に同時に処理ができる上限値を決めたくなることがあります。
たとえば、某拝承系SIerが開発しているCosminexusではURL毎に同時アクセスの最大数を設定でき、同時アクセス数を超えた場合はキューに溜めるという機能が存在します。
http://itdoc.hitachi.co.jp/manuals/link/cosmi_v0870/APSE/EU030187.HTM

このようなエンタープライズなアプリケーション・サーバーを使わなくてもJavaではjava.util.concurrent.Semaphoreを使用することで、単純なアクセス制限であれば簡単に作成できるので、実装したいと思います。

java.util.Semaphoreとは

同時アクセス数の制限をするために使用します(ﾄﾞﾔｧ

・・・・さて、以下の様な感じで使用できます。

newするときにSemaphoreが持つリソース量の初期値を決める。
各処理は重み付けをSemaphoreに伝え、Semaphoreからリソースを受け取る。Semaphoreは払いだしたリソースの分を自分の持っているリソース量を減らす。
Semaphoreがリソースを払い出せない場合は、リソースが払い出せるようになるまで待つ。
処理が終わったらSemaphoreにリソースを返却する。返却することでSemaphoreは自分が持ってるリソース量を増やす。

コードで書くとこんな感じです。

//
//クラス変数
//
//ここはすべてのスレッドからアクセスできるようにstaticで分離しておく。
static Semaphore s = new Semaphore(1000);

//
//メソッド内の処理
//
int throughput = resource.getThroughput();
//リソースを払いだしてもらう
s.acquire(throughput);
try {
//実処理を行う
    resource.access();
} finally {
//処理が終わったらリソースを返却する
    s.release(throughput);
}

注意点としては、どうもSemaphoreのreleaseは初期値以上にもできてしまうらしいので、finally以外でreleaseするとかすると微妙な感じになる可能性があることかしら。
詳しくは以下でも。
http://d.hatena.ne.jp/torutk/20120226/p1

Semaphoreを複数リソースに適用する。

みんな大好きjava.util.ConcurrentHashMapを使って以下の様にします。

//
// リソースをロックするためのヘルパークラス
//
class ResourceLock {

    public static Map<String, Semaphore> lockMap = new ConcurrentHashMap<>();

    public static void acquire(Resource resource) throws InterruptedException {
        //単純化のため、リソース量の総量については1024で固定しています。
        lockMap.computeIfAbsent(resource.getResourceId(), s -> new Semaphore(1024)).acquire(resource.getThroughput());
    }

    public static void release(Resource resource) throws InterruptedException {
        lockMap.get(resource.getResourceId()).release(resource.getThroughput());
    }
}
//
//使い方
//
ResourceLock.acquire(resource);
try {
    resource.access();
} finally {
    ResourceLock.release(resource);
}

リソース毎にSemaphoreがなければ新しく作る、Semaphoreがなければ既存のやつを使用する、その後にリソースを分捕り返却する。簡単だ。
詳しくは以下でも
http://www.slideshare.net/makingx/jjug

リソースの種類が無数にある場合

上の実装だと、ConcurrentHashMapでSemaphoreをずっと持ち続けているため、リソースの種類がたくさんある場合はリソース毎にSemaphoreを持つためメモリ使用量が酷いことになります。*1
そのため、アクセスがないものについては消してしまうように実装を変更します。
消すという処理を新しく書くのも面倒くさいので、commons-collectionsにあるorg.apache.commons.collections4.map.LRUMapを使用します。

class ResourceLock {

    //LRUMapはスレッドセーフでないため、ロックが掛かってしまうが仕方がないとする
    //LRUMapの初期値は直近で持っておくMapの要素数
    public static Map<String, Semaphore> lockMap = Collections.synchronizedMap(new LRUMap<>(1024));

    public static void acquire(Resource resource) throws InterruptedException {
        lockMap.computeIfAbsent(resource.getResourceId(), s -> new Semaphore(1024)).acquire(resource.getThroughput());
    }

    public static void release(Resource resource) throws InterruptedException {
        lockMap.get(resource.getResourceId()).release(resource.getThroughput());
    }
}

リソースの種類が無数にあって、さらにアクセス数もひどい場合

LRUMapのアクセスに対してはロックが掛かってしまうのですが、すべてのスレッドからのアクセスで全体のロックをかけるというのは非効率的なので、ConcurrentHashMapを使用してロックをかける範囲を限定的にします。

class ResourceLock {

    public static Map<Integer, Map<String, Semaphore>> lockMap = new ConcurrentHashMap<>();

    public static void acquire(Resource resource) throws InterruptedException {
        //リソースのIDでロックをかける範囲を分散させる
        //今回はハッシュコードを利用して8分割している
        lockMap.computeIfAbsent(resource.getResourceId().hashCode() % 8,
                s -> Collections.synchronizedMap(new LRUMap<String, Semaphore>(128))).computeIfAbsent(resource.getResourceId(),
                        a -> new Semaphore(1024)).acquire(resource.getThroughput());
    }

    public static void release(Resource resource) throws InterruptedException {
        lockMap.get(resource.getResourceId().hashCode() % 8).get(resource.getResourceId()).release(resource.getThroughput());
    }
}

8分割したことで、全体でリソースへのロックを掛けるのを共有するわけではなくなり、ずいぶんと性能が改善したと思います。

終わり。

*1:数百万とか数千万種類とかある場合