桃缶食べたい。

Akka Stream の Graph の構成要素

Akka Stream の依存性設定

<!-- Akka Stream -->
<dependency>
  <groupId>com.typesafe.akka</groupId>
  <artifactId>akka-stream_2.12</artifactId>
  <version>2.5.9</version>
</dependency>

Akka Stream を動かしてみる

import java.util.concurrent.CompletionStage;

import akka.Done;
import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.stream.ActorMaterializer;
import akka.stream.Materializer;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;

public class AkkaStreamSample01 {

    public static void main(String[] argv) {
        // Akka Stream の処理系を用意する
        final ActorSystem system = ActorSystem.create("AkkaStreamSample");
        final Materializer materializer = ActorMaterializer.create(system);

        // Akka Stream で処理する Graph を用意する

        // Source は 1 〜 100 の Integer を生み出す
        final Source<Integer, NotUsed> source = Source.range(1, 100);
        // Sink は標準出力へ受け取った値を出力する
        final Sink<Integer, CompletionStage<Done>> sink = Sink.foreach(i -> System.out.println(Thread.currentThread().getId() + ": " + i));

        // Source を Sink につなげて Graph にする
        final RunnableGraph<NotUsed> graph = source.to(sink);

        // Akka Stream の処理系を使って、実際に Graph に定義された処理を実行する
        final NotUsed done = graph.run(materializer);

        // 実行後には Akka Stream の処理系を終了する
        system.terminate();
    }

}

    public static void main(String[] argv) {
        // Akka Stream の処理系を用意する
        final ActorSystem system = ActorSystem.create("AkkaStreamSample");
        final Materializer materializer = ActorMaterializer.create(system);

        // Akka Stream で処理する Graph を用意する

        // Source は 1 〜 100 の Integer を生み出す
        final Source<Integer, NotUsed> source = Source.range(1, 100);

        // Akka Stream の処理系を使って、実際に Graph に定義された処理を実行する
        // Sink は標準出力へ受け取った値を出力する
        // Source を Sink につないだことで得られる RunnableGraph に対して run() を実行し、実処理を行う
        final CompletionStage<Done> done = source.runWith(Sink.foreach(i -> System.out.println(Thread.currentThread().getId() + ": " + i)), materializer);

        // 実行後には Akka Stream の処理系を終了する
        done.thenRun(() -> system.terminate());
    }

Graph をまとめて別の Graph にする

import java.math.BigInteger;
import java.nio.file.Paths;
import java.util.concurrent.CompletionStage;

import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.stream.ActorMaterializer;
import akka.stream.IOResult;
import akka.stream.Materializer;
import akka.stream.javadsl.FileIO;
import akka.stream.javadsl.Flow;
import akka.stream.javadsl.Keep;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;
import akka.util.ByteString;

public class AkkaStreamSample02 {

    public static void main(String[] argv) {
        // Akka Stream の処理系を用意する
        final ActorSystem system = ActorSystem.create("AkkaStreamSample");
        final Materializer materializer = ActorMaterializer.create(system);

        // すでに作成済みの Source に処理を追加して新たな Source を作成する

        // 1 〜 100 の Integer を生み出す Source を作成する
        final Source<Integer, NotUsed> otherSource = Source.range(1, 100);
        // 1 ~ 100 の Integer を生み出す Source を使って、100 個の 2 の累乗を生み出す Source を作成する
        final Source<BigInteger, NotUsed> source = otherSource.map(n -> BigInteger.valueOf(2).pow(n));


        // 複数の Flow をつなぎ合わせた Sink を新たな Sink として作成する

        final Sink<BigInteger, CompletionStage<IOResult>> sink = Flow
                // BigInteger 型のデータを処理する Flow として Sink を作成する
                .of(BigInteger.class)
                // 受け取ったデータを 1 行ずつ出力するために、1 行分の文字列に変換する Flow を通す
                .map(s -> s.toString() + "\n")
                // 受け取ったデータをファイル出力を行う Sink につなげるために ByteString 型に変換する Flow を通す
                .map(n -> ByteString.fromString(n.toString()))
                // ファイル出力を行う Sink につなぐ
                .toMat(FileIO.toPath(Paths.get("result.txt")), Keep.right());


        // 作成済みの Sink と Flow から Graph を作成する

        final CompletionStage<IOResult> result = source
                // 受け取った数値のうち、1024 でない数に絞り込む Flow を通す
                .filter(i -> !i.equals(BigInteger.valueOf(1024)))
                // ファイル出力する Sink につないで Graph を実行する
                .runWith(sink, materializer);

        // Graph に対して run() を実行し、実処理を開始する
        result.thenRun(() -> system.terminate());
    }

}