Javaではじめるakka入門

はじめまして。
ブロググループ所属の見原と申します。
今回は先日担当した案件で利用した、akkaを紹介させていただきます。

はじめに

akkaはtypesafeが提供する、イベント駆動の分散並列型アプリケーションフレームワークです。
並列処理が簡単に記述出来るほか、複数のマシンを用いた分散処理の実現、
「let it crash」という設計思想に基づいたロジックとリカバリ処理の分離などが特徴です。
akkaのサンプルはscalaが多いのですが、今回はjavaで説明していきます。

まずはMavenを利用したアプリケーションを作成します。
akkaを利用するため、以下をpom.xmlに記載します。

<repositories>
    <repository>
        <id>typesafe</id>
        <name>Typesafe Repository</name>
        <url>http://repo.typesafe.com/typesafe/releases/</url>
    </repository>
</repositories>
<properties>
    <com.typesafe.akka.version>2.1.4</com.typesafe.akka.version>
</properties>   

<dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-actor_2.10</artifactId>
    <version>${com.typesafe.akka.version}</version>
</dependency>
<dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-remote_2.10</artifactId>
    <version>${com.typesafe.akka.version}</version>
</dependency>
<dependency>
    <groupId>com.typesafe.akka</groupId>
    <artifactId>akka-kernel_2.10</artifactId>
    <version>${com.typesafe.akka.version}</version>
</dependency>

メッセージの送信

まずは一番基本的なメッセージの送信を試してみます。

public class SimpleActorSystem {

    public static void main(String[] args) {
        ActorSystem system = ActorSystem.create("system");
        ActorRef ref = system.actorOf(new Props(SimpleActor.class), "simpleActor");

        String message = "hello.";
        ref.tell(message, null);

        try {
            Thread.sleep(3000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        system.shutdown();

        System.out.println("end.");
    }
}

4行目で生成しているActorSystemはakkaにおいて中核となるクラスです。
次の行では、これを用いて実際にメッセージの処理をするActorを生成しています。
メッセージの送信にはtellメソッドを利用します。
第一引数には送信したいメッセージを、第二引数には送信元とするActorを指定します。
今回は送信先で特に送信元に対して処理を行わないのでnullを指定しています。

次に、メッセージを受け取るActorを実装します。

public class SimpleActor extends UntypedActor {

    @Override
    public void onReceive(Object message) throws Exception {
        if (message instanceof String) {
            System.out.println("message:" + message);
        } else {
            System.out.println("unhandled message.");

            // 想定しない型のメッセージはスルーする
            unhandled(message);
        }
    }

    @Override
    public void preStart() {
        System.out.println("preStart");
    }

    @Override
    public void postStop() {
        System.out.println("postStop");
    }

}

Actorは基本的にUntypedActorを継承します。
onReceiveメソッドが送信されたメッセージを処理するメソッドになります。
今回はStringのメッセージを出力、それ以外のメッセージはスルーするつくりになっています。
preStartはActorのスタート時に、postStopはActorの停止時に実行されるので、
何らかの処理をしたいときに任意にOverrideします。

これで実行の準備が整いました。
SimpleActorSystemを実行してみると、以下のような結果が得られます。

preStart
message:hello.
end.
postStop

メッセージの送受信

次に、メッセージを送ってActorの返事を待ってみます。
先ほどtellしていた部分を以下のように書き換えます。

private Integer result = 0;
try {
    // メッセージを送信し、結果を受け取る
    result = (Integer) Await.result(Patterns.ask(actor, message, 5000), Duration.create(5000, TimeUnit.MILLISECONDS));
} catch (Exception e) {
    e.printStackTrace();
}
System.out.println("result=" + result);

try {
    result = (Integer) Await.result(Patterns.ask(actor, message, 5000), Duration.create(5000, TimeUnit.MILLISECONDS));
} catch (Exception e) {
    e.printStackTrace();
}
System.out.println("result=" + result);

try {
    // 想定しない型を送信する。Exceptionが発生する
    result = (Integer) Await.result(Patterns.ask(actor, 1, 5000), Duration.create(5000, TimeUnit.MILLISECONDS));
} catch (Exception e) {
    e.printStackTrace();
}
System.out.println("result=" + result);

メッセージの送信にはtellを使用していましたが、代わりにPatternsのaskメソッドを利用します。
こちらを用いると結果がFutureで返ってくるので、Awaitのresultメソッドを用いてその結果の待機、取得を行います。

次に、Actorの中身を以下のように書き換えます。

int state = 0;

@Override
public void onReceive(Object message) throws Exception {
    if (message instanceof String) {
        System.out.println("message:" + message);
        sender().tell(++state, self());
    } else {
        System.out.println("unhandled message.");
        unhandled(message);
    }
}

先ほどの例では意図するメッセージを受け取ったとき、それを出力するだけでした。
しかし、今回は送信元に対してメッセージを送り返しています。

こちらを実行すると以下のような結果になります。

preStart
message:hello.
result=1
message:hello.
result=2
unhandled message.
result=2
java.util.concurrent.TimeoutException: Futures timed out after [5000 milliseconds]
    at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:96)
    at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:100)
    at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:107)
    at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53)
    at scala.concurrent.Await$.result(package.scala:107)
    at scala.concurrent.Await.result(package.scala)
    at jp.ameba.blog.akka.sample.ask.AskActorSystem.main(AskActorSystem.java:43)
end.
postStop

2回目の送信までは正常に動作しています。
しかし、3回目の送信ではActorが意図しない型を送信しているため、
Actorが送信元にメッセージを送り返してくれず、タイムアウトが発生しています。

複数サーバを用いた分散処理

先頭でもお話しした通り、akkaを用いると、複数のマシンを用いた分散処理の実現が容易に可能です。
最後に、こちらの実装をご紹介したいと思います。

今回はローカル実行用とリモート実行用にActorSystemを二つ用意します。
まずはリモート実行用のクラスを作成します。

public class RemoteSystem implements Bootable {
    final ActorSystem system = ActorSystem.create("remoteSys", ConfigFactory.load().getConfig("remoteSetting"));

    @Override
    public void startup() {
        System.out.println("start");
    }
   
    @Override
    public void shutdown() {
        system.shutdown();
    }
}

簡単ですね！
これまでと異なるのは、Bootableインターフェースを実装していることです。
startupメソッドとshutdownメソッドを実装してあげる必要があります。
さらにもう一点異なるのは、ActorSystemの生成の際、設定をファイルから読み込んでいることです。
設定ファイルから「remoteSetting」という設定を探して読み込んでいます。

設定ファイルは「application.conf」という名前で以下のように記述します。

localSetting {
    include "common"
    akka {
        remote {
            transport = "akka.remote.netty.NettyRemoteTransport"
            netty {
                hostname = "127.0.0.1"
                port = 2552
            }
        }
    }
}

remoteSetting {
    include "common"
    akka {
        remote {
            transport = "akka.remote.netty.NettyRemoteTransport"
            netty {
                hostname = "127.0.0.1"
                port = 2553
            }
        }
    }
}

ローカルで実行するための設定と、先ほど読み込んだリモート用の設定を記述してあります。
実際に複数サーバで実行したいのですが、今回はテストのために同じIP、異なるポートを指定してひとつのマシンで実行出来るようにしています。

また、includeという記述がありますが、これはほかのファイルに記述したものを読み込むものです。
今回、「common」をincludeするという記述になっているので、
「common.conf」という名称のファイルを探して読み込みます。
common.confの内容は以下のようになっています。

akka {
    actor {
        provider = "akka.remote.RemoteActorRefProvider"
    }
}

次にローカル用のクラスを生成します。

public class LocalSystem implements Bootable {

    final ActorSystem system = ActorSystem.create("localSys", ConfigFactory
            .load().getConfig("localSetting"));

    private static final String AKKA_PROTOCOL = "akka";
    private static final String REMOTE_SYSTEM_NAME = "remoteSys";
    private static final String REMOTE_HOST_IP = "127.0.0.1";
    private static final int REMOTE_HOST_PORT = 2553;

    @Override
    public void startup() {
        String message = "hello.";

        // ローカルのActorを生成
        System.out.println("local actor test.");
        ActorRef localActor = system.actorOf(new Props(RemoteActor.class),
                "localActor");
        localActor.tell(message, null);

        try {
            Thread.sleep(2000L);
        } catch (InterruptedException e) {
        }
        System.out.println("-----");

        // リモートのActorを生成
        System.out.println("remote actor test.");
        ActorRef remoteActor = system.actorOf(new Props(RemoteActor.class)
        .withDeploy(new Deploy(new RemoteScope(new Address(
                AKKA_PROTOCOL, REMOTE_SYSTEM_NAME, REMOTE_HOST_IP,
                REMOTE_HOST_PORT)))), "remoteActor");
        remoteActor.tell(message, null);

        try {
            Thread.sleep(2000L);
        } catch (InterruptedException e) {
        }
        System.out.println("-----");

        // ラウンドロビンでActorにメッセージを送信
        System.out.println("router test with RoundRobinRouter.");
        List<ActorRef> actors = new ArrayList<>();
        actors.add(remoteActor);
        actors.add(localActor);
        ActorRef roundRobinRouter = system.actorOf(new Props(RemoteActor.class)
        .withRouter(RoundRobinRouter.create(actors)));
        for (int i = 0; i < 6; i++) {
            String roundRobinMessage = message + i;
            roundRobinRouter.tell(roundRobinMessage, null);
        }

        try {
            Thread.sleep(2000L);
        } catch (InterruptedException e) {
        }
        System.out.println("-----");

        system.shutdown();

        System.out.println("end.");
    }
}

ローカル用のActorの生成はこれまでと一緒です。
リモート用のActorはDeployというクラスを用いて生成しています。
純粋にメッセージを送るだけなら、生成したActorに対してtellを実行するだけで可能です。
しかし、実際に利用する場合、特定のサーバに存在するActorに対してメッセージを送信するのではなく、
複数サーバに存在するActorに対して平等にメッセージを送信することで、負荷の分散を図りたいものです。

そこで便利なのがRouterです。
今回はRoundRobinRouterを利用しています。
このルーターに対してメッセージの送信を行うと、このルーターが管理しているActorに対して、
名前の通りラウンドロビンでメッセージの送信を行ってくれます。
このほかにも、ランダムでメッセージの送信を行うRandomRouterや、
全Actorに一括送信を行うBroadcastRouterなどが存在します。

では、実際に今記述したものをテストしてみます。
Bootstrapインターフェースを実装したクラスは、akkaの提供するスクリプトで実行することができます。

http://typesafe.com/platform/runtime/akka/download
こちらより、最新のakkaをダウンロードして解凍します。

作成したものをjarにして、deployディレクトリに配置すれば準備完了です。
linuxならbinディレクトリにあるakka、Windowsならakka.batに対して実行したいBootableを実装したクラスを完全限定名で指定すれば、プロセスが起動します。

まず、リモート用のActorSystemを起動してみます。

akka-2.1.4\bin>akka.bat jp.ameba.blog.akka.sample.remote.RemoteSystem
Starting Akka...
Running Akka 2.1.4
Deploying file:/akka-2.1.4/bin/../deploy/akka-s
ample-app.jar
[INFO] [07/08/2013 11:27:35.109] [main] [NettyRemoteTransport(akka://remoteSys@127.0.0.1:2553)] RemoteServerStarted@akka://remoteSys@127.0.0.1:2553
Starting up jp.ameba.blog.akka.sample.remote.RemoteSystem
start
Successfully started Akka

次に同じようにローカル用のActorSystemを起動し、メッセージを送信します。
ローカルの実行結果は以下の通りです。

akka-2.1.4\bin>akka.bat jp.ameba.blog.akka.sample.remote.LocalSystem
(略)
local actor test.
static initialization completed. hostName=pctest
message=hello., state=1, hostName=pctest
-----
remote actor test.
[INFO] [07/08/2013 11:35:37.186] [main] [NettyRemoteTransport(akka://localSys@127.0.0.1:2552)] RemoteClientStarted@akka://remoteSys@127.0.0.1:2553
-----
router test with RoundRobinRouter.
message=hello.1, state=2, hostName=pctest
message=hello.3, state=3, hostName=pctest
message=hello.5, state=4, hostName=pctest
-----
Successfully started Akka

Shutting down Akka...
Shutting down jp.ameba.blog.akka.sample.remote.LocalSystem
Successfully shut down Akka

次にリモートの実行結果は以下の通りです。

[INFO] [07/08/2013 11:35:37.232] [remoteSys-10] [NettyRemoteTransport(akka://remoteSys@127.0.0.1:2553)] RemoteClientStarted@akka://localSys@127.0.0.1:2552
static initialization completed. hostName=pctest
message=hello., state=1, hostName=pctest
message=hello.0, state=2, hostName=pctest
message=hello.2, state=3, hostName=pctest
message=hello.4, state=4, hostName=pctest
[INFO] [07/08/2013 11:35:45.333] [remoteSys-7] [NettyRemoteTransport(akka://remoteSys@127.0.0.1:2553)] RemoteClientShutdown@akka://localSys@127.0.0.1:2552

RoundRobinRouterに対して送信したメッセージがローカルとリモートのActorに対して均等に送信されていることが確認できます。

まとめ

今回はakkaの初歩の部分のみのご紹介になりましたが、
他にもDispatcherを用いると並列処理が容易に記述できたり、
SupervisorStrategyと用いるとロジックとリカバリ処理の分離が出来たりと、
様々な便利な機能を有しています。
また、まだテスト段階ではありますが、将来的にはクラスタリングもサポートされます。

この記事でご紹介したものと、SupervisorStrategyなどを含めたサンプルアプリは以下に公開しておりますので、
ご興味のある方はご覧ください。
https://github.com/tm8r/akka_sample_app

サイバーエージェント公式エンジニアブログ

CyberAgent Engineer Blog.

Javaではじめるakka入門

はじめに

メッセージの送信

メッセージの送受信

複数サーバを用いた分散処理

まとめ