リアクティブストリームの紹介-Java開発者向け

こんにちは、Habr！

今日は、すばらしい本「 Reactive Design Patterns 」で取り上げられているトピックの1つに戻ります。 Akka Streamsとストリーミングデータ全般についてお話します-ローランド・クーンの本では、10章と15章から17章がこれらの問題に当てられています。

Jetストリームは、データを非同期的にストリーミングする標準的な方法です。 これらはjava.util.concurrent.FlowインターフェースとしてJava 9に含まれていましたが、現在ではさまざまなアプリケーションでストリーミングコンポーネントを作成するための実際の命の恩人になりつつあります。この配置は今後も続くでしょう。 リアクティブストリームは「まさに」標準であり、それだけでは価値がないことに注意してください。 実際には、この標準の1つまたは別の特定の実装が使用されます。今日は、Akka Streamsについて説明します。AkkaStreamsは、ジェットストリームの最初の実装の1つです。

コンテキスト

典型的なストリーム処理パイプラインはいくつかのステップで構成され、各情報は次のステップに（つまり降順で）送信されます。 したがって、2つの隣接する手順を実行して、親をサプライヤ、次のステップをデータコンシューマと見なすと、サプライヤはコンシューマよりも遅く、またはコンシューマよりも速く動作できることがわかります。 サプライヤの作業が遅くなると、すべては問題ありませんが、消費者がサプライヤと歩調を合わせないと状況は複雑になります。 この場合、消費者は慎重に処理するために（できる限り）必要なデータであふれることがあります。

過剰なデータに対処する最も簡単な方法は、処理できないすべてのものを取得して破棄することです。 これは、たとえばネットワーク機器を操作する場合など、まさに彼らが行うことです。 しかし、何もドロップしたくない場合はどうなりますか？ その後、背圧が役立ちます。

バックプレッシャの概念は、リアクティブストリームのコンテキストでは非常に重要であり、パイプラインの隣接するリンク間で転送されるデータの量を制限するため、リンクはオーバーフローしません。 リアクティブアプローチの最も重要な側面は、絶対に必要でない限りブロッキングを防ぐことなので、リアクティブストリームでの背圧の実装もノンブロッキングでなければなりません。

どうやって

Reactive Streams標準では、多数のインターフェースが定義されていますが、そのような実装は定義されていません。 これは、org.reactivestreams：Reactive-streamsに依存関係を追加するだけで、その場で踏みつけるだけで、特定の実装が必要であることを意味します。 Reactive Streamsには多くの実装がありますが、この記事ではAkka Streamsと対応するJavaベースのDSLを使用します。 他の実装には、 RxJava 2.xまたはReactorなどが含まれます。

使用例

新しいCSVファイルを追跡し、各ファイルをストリーミングベースで処理し、その場で集計を実行し、収集した結果をWebソケットに（リアルタイムで）送信するディレクトリがあるとします。 さらに、到達すると電子メール通知がトリガーされる集約データの蓄積に特定のしきい値を設定します。

この例では、CSV行にはペア（ id 、 value ）が含まれ、 idは2行ごとに変更されます。次に例を示します。

370582,0.17870700247256666
370582,0.5262255382633264
441876,0.30998025265909457
441876,0.3141591265785087
722246,0.7334219632071504
722246,0.5310146239777006

共通のIDを持つ2行の平均値を計算し、それが0.9を超える場合にのみWebソケットに送信します。 さらに、5番目の値ごとにWebソケットに到着した後に電子メール通知を送信したいと思います。 最後に、Webソケットから受信したデータを読み取り、表示します。これは、JavaScriptで作成された簡単なフロントエンドを介して行われます。

建築

Akkaエコシステムの多数のツールを使用します（図1を参照）。 当然、Akka Streamsはシステム全体の中心に位置し、ストリーミングベースでリアルタイムにデータを処理できます。 Alpakkaを使用してCSVファイルを読み取ります。これは、Akka Streamsをさまざまなテクノロジー、プロトコル、またはライブラリと統合するためのコネクタのセットです。 Akka Streamsはリアクティブフローであるため、Alpakkaエコシステム全体が他のRS実装にも利用できることは興味深いことです。相互運用性を実現するように設計されているのはこのようなRSインターフェイスです。 最後に、Akka HTTPを使用してWebソケットのエンドポイントを提供します。 この場合の最良の部分は、Akka HTTPがAkka Streamsとシームレスに統合されることです（実際、「裏側」を使用します）。したがって、Webソケットとしてストリームを提供することは難しくありません。



図 1.アーキテクチャの概要

このスキームを従来のJava EEアーキテクチャと比較すると、ここではすべてがはるかに単純であることに気付くでしょう。 コンテナとBeanはありませんが、単純なスタンドアロンアプリケーションです。 さらに、Java EEスタックはストリーミングアプローチをまったくサポートしていません。

Akka Streamsの基本

Akka Streamsでは、処理パイプライン（グラフ）は、 Source （ソース）、 Sink （トラップ）、 Flow （処理ステップ）の3つの要素で構成されています。

これらのコンポーネントに基づいて、本質的にデータ処理の単なるレシピであるグラフを定義します。 計算は行われません。 パイプラインが機能するためには、グラフを具体化する、つまり実行可能な形式にする必要があります。 これを行うには、グラフの定義を最適化し、最終的にそれを実行する、いわゆるマテリアライザーが必要になります。 ただし、組み込みのActorMaterializerは実質的に競合していないため、他の実装を使用することはほとんどありません。
コンポーネントのタイプのパラメーターをよく見ると、各コンポーネント（対応するタイプの入力/出力を除く）に不思議なタイプのマットがあることがわかります。 これは、いわゆる「実体化された値」を指します-これは、グラフの外部からアクセス可能な値です（グラフのステップ間の内部通信にのみ使用可能な入力/出力のタイプとは対照的に、図2を参照）。 実体化された値を無視したい場合（そして、これはグラフのステップ間でデータを転送することだけに関心がある場合によく起こります）、このオプションを示すNotUsedタイプの特別なパラメーターがあります。 JavaのVoidと比較できますが、意味的には少しロードされます。「この値を使用しない」という意味では、 Voidよりも情報量が多くなります。 また、一部のAPIは同様のタイプの完了を使用し、特定のタスクが完了したことを通知します。 おそらく、これらの両方の場合の他のJavaライブラリはVoidを使用しますが、Akka Streamsでは、すべてのタイプが有用なセマンティクスで最大を埋めようとします。



図 2.タイプFlowのパラメーターの説明

アプリ

次に、CSVハンドラーの特定の実装に移りましょう。 まず、Akka Streamsグラフを定義し、次にAkka HTTPプロトコルを使用して、ストリームをWebソケットに接続します。

ストリームコンベヤーのコンポーネント

ストリーミングパイプラインの入力ポイントで、目的のディレクトリに新しいCSVファイルが出現したかどうかを追跡します。 これにはjava.nio.file.WatchServiceを使用しますが、ストリーミングアプリケーションがあるため、イベントのソース（ Source ）を取得して操作する必要があり、コールバックを通じてすべてを整理する必要はありません。 幸いなことに、このようなSourceは、 DirectoryChangesSourceコネクタの1つの形式でAlpakkaで既に利用可能ですalpakka-file一部であり、 WatchService 「 WatchService 」で使用されalpakka-file 。

 private final Source<Pair<Path, DirectoryChange>, NotUsed> newFiles = DirectoryChangesSource.create(DATA_DIR, DATA_DIR_POLL_INTERVAL, 128); 

したがって、タイプPair<Path, DirectoryChange>要素を提供するソースを取得します。 新しいCSVファイルのみを選択し、「ダウン」して転送するように、それらをフィルタリングします。 このデータ変換と後続のすべてのデータ変換では、Flowと呼ばれる小さな要素を使用して、本格的な処理パイプラインを形成します。

 private final Flow<Pair<Path, DirectoryChange>, Path, NotUsed> csvPaths = Flow.<Pair<Path, DirectoryChange>>create() .filter(this::isCsvFileCreationEvent) .map(Pair::first); private boolean isCsvFileCreationEvent(Pair<Path, DirectoryChange> p) { return p.first().toString().endsWith(".csv") && p.second().equals(DirectoryChange.Creation); } 

たとえば、汎用のcreate()メソッドを使用してFlowを作成できます。これは、入力タイプ自体が汎用の場合に役立ちます。 ここで、結果のストリームは、 DATA_DIR現れるすべての新しいCSVファイルを（ Pathの形式で）生成します。

次に、各ファイルのストリームによって選択されたパスを文字列に変換します。 ソースを別のソースに変換するには、 flatMap*メソッドのいずれかを使用できます。 どちらの場合も、各入力要素からSourceを作成し、結果のソースのいくつかを何らかの形で組み合わせて、新しい単一のリンクソースソースを結合します。 この場合、同じid持つ行が隣同士に残るように行の順序を保持するため、 flatMapConcatに焦点を当てます。 Pathをバイトストリームに変換するには、組み込みのFileIOユーティリティを使用します。

 private final Flow<Path, ByteString, NotUsed> fileBytes = Flow.of(Path.class).flatMapConcat(FileIO::fromPath); 

今回はof()メソッドを使用して新しいストリームを作成します-入力タイプが一般化されていない場合に便利です。

上記のByteStringは、Akka Streamsで採用されているバイトシーケンス表現です。 この場合、バイトストリームをCSVファイルとして解析します。このために、Alpakkaモジュールの1つ、今回はalpakka-csvを再び使用します。

 private final Flow<ByteString, Collection<ByteString>, NotUsed> csvFields = Flow.of(ByteString.class).via(CsvParsing.lineScanner()); 

ここで使用されるviaコンビネータに注意してください。これにより、グラフの別のステップ（ Sourceまたは別のFlow ）で受信した出力に任意のFlowを添付できます。 結果は、CSVファイルの1行のフィールドにそれぞれ対応する要素のストリームです。 次に、それらを対象領域のモデルに変換できます。

 class Reading { private final int id; private final double value; private Reading(int id, double value) { this.id = id; this.value = value; } double getValue() { return value; } @Override public String toString() { return String.format("Reading(%d, %f)", id, value); } static Reading create(Collection<ByteString> fields) { List<String> fieldList = fields.stream().map(ByteString::utf8String).collect(toList()); int id = Integer.parseInt(fieldList.get(0)); double value = Double.parseDouble(fieldList.get(1)); return new Reading(id, value); } } 

そのように変換するには、 mapメソッドを使用して、 Reading.createメソッドへのリンクをReading.createます。

 private final Flow<Collection<ByteString>, Reading, NotUsed> readings = Flow.<Collection<ByteString>>create().map(Reading::create); 

次の段階では、読み取り値をペアで追加し、各グループの平均値を計算し、特定のしきい値に達した場合にのみさらに情報を送信する必要があります。 平均を非同期的に計算する必要があるため、 mapAsyncUnorderedメソッドを使用します。 mapAsyncUnorderedメソッドは、指定されたレベルの並列性で非同期操作を実行します。

 private final Flow<Reading, Double, NotUsed> averageReadings = Flow.of(Reading.class) .grouped(2) .mapAsyncUnordered(10, readings -> CompletableFuture.supplyAsync(() -> readings.stream() .map(Reading::getValue) .collect(averagingDouble(v -> v))) ) .filter(v -> v > AVERAGE_THRESHOLD); 

上記のコンポーネントを定義したら、統合されたコンベヤーをそれらから追加する準備ができました（コンビネーターviaて使い慣れたものvia使用）。 それはまったく複雑ではありません：

 private final Source<Double, NotUsed> liveReadings = newFiles .via(csvPaths) .via(fileBytes) .via(csvFields) .via(readings) .via(averageReadings); 

ご注意

上記のようにコンポーネントを組み合わせる場合、コンパイラは、互換性のないデータ型を含む2つのブロックを誤って接続しないようにして、私たちを保護します。

Webソケットとしてのストリーム

次に、Akka HTTPを使用して、このような役割を果たす単純なWebサーバーを作成します。

• 読み取りソースをWebソケットとして提供し、
• Webソケットに接続し、受信したデータを表示する簡単なWebページを発行します。

Akka HTTPを使用してWebサーバーを作成する費用はかかりませんHttpAppを継承し、DSLルートで必要なマッピングを提供するだけです。

 class Server extends HttpApp { private final Source<Double, NotUsed> readings; Server(Source<Double, NotUsed> readings) { this.readings = readings; } @Override protected Route routes() { return route( path("data", () -> { Source<Message, NotUsed> messages = readings.map(String::valueOf).map(TextMessage::create); return handleWebSocketMessages(Flow.fromSinkAndSourceCoupled(Sink.ignore(), messages)); } ), get(() -> pathSingleSlash(() -> getFromResource("index.html") ) ) ); } } 

ここで2つのルートが定義されています： /data 、つまりWebソケットのエンドポイント、および/それに沿って簡単なフロントエンドが発行されます。 Akka StreamsのSourceをWebソケットのエンドポイントとして提供するのがどれほど簡単かはすでに明らかですhandleWebSocketMessagesをhandleWebSocketMessages 、そのタスクはWebソケットへの接続へのHTTP接続を改善し、そこに着信および発信データが処理されるストリームを編成することです

WebSocketストリームとしてモデル化されています。つまり、発信メッセージと着信メッセージがクライアントに送信されます。 この場合、着信データを無視して、「着信」側がSink.ignore()れているストリームを作成します。 Webソケットハンドラーストリームのアップストリーム側は、平均値の取得元であるソースに単純に接続されています。 平均値が表される形式のdouble数値で行う必要があるのは、それぞれをTextMessageに変換することです。これは、Akka HTTPでWebソケットデータに使用されるラッパーです。 すべては、すでにおなじみのmapメソッドを使用して行われます。

サーバーを起動するには、ホスト名とポートを指定してstartServerメソッドを起動するだけです。

 Server server = new Server(csvProcessor.liveReadings); server.startServer(config.getString("server.host"), config.getInt("server.port")); 

フロントエンド

Webソケットからデータを受信して​​表示するには、受信した値をtextareaに添付するだけの完全にシンプルなJavaScriptコードを使用します。 このコードはES6構文を使用しており、最新のブラウザーで正常に機能するはずです。

 let ws = new WebSocket("ws://localhost:8080/data"); ws.onopen = () => log("WS connection opened"); ws.onclose = event => log("WS connection closed with code: " + event.code); ws.onmessage = event => log("WS received: " + event.data); 

logメソッドはメッセージをtextareaに添付し、タイムスタンプも付けます。

打ち上げ

アプリケーションを実行してテストするには、次のものが必要です。

• サーバーを起動します（ sbt run ）、
• ブラウザでlocalhost ：8080に移動します（デフォルトを変更した場合は、選択したホスト/ポートに移動します）。
• src/main/resources/sample-dataからプロジェクトルートのdataディレクトリに1つまたは複数のファイルをコピーします（構成でcsv-processor.data-dir変更しなかった場合）。
• サーバーログおよびブラウザでのデータの表示方法を確認してください。

メールトリガーを追加する

アプリケーションの最後の仕上げは、5番目の要素がすべてWebソケットに到着した後に送信される電子メール通知をシミュレートするサイドチャネルです。 基本的な要素の伝達を妨げないように、「横向き」に機能する必要があります。

この動作を実装するには、Akka Streamsのより高度な機能であるGraph DSL言語を使用します。この機能では、独自のグラフステップを記述し、フローが2つの部分に分岐します。 最初は単純に値をWebソケットに送信し、2番目は次の5秒の有効期限を制御し、電子メールで通知を送信します。図を参照してください。 3。



図 3.メールを送信する独自​​のグラフステップ

Broadcast組み込みステップを使用します。このステップでは、発表された一連の結論に入力が送信されます。 独自のトラップMailerも作成します。

 private final Graph<FlowShape<Double, Double>, NotUsed> notifier = GraphDSL.create(builder -> { Sink<Double, NotUsed> mailerSink = Flow.of(Double.class) .grouped(EMAIL_THRESHOLD) .to(Sink.foreach(ds -> logger.info("Sending e-mail") )); UniformFanOutShape<Double, Double> broadcast = builder.add(Broadcast.create(2)); SinkShape<Double> mailer = builder.add(mailerSink); builder.from(broadcast.out(1)).toInlet(mailer.in()); return FlowShape.of(broadcast.in(), broadcast.out(0)); }); 

GraphDSL.create()メソッドを使用して独自のグラフステップを作成します。このメソッドでは、グラフBuilderインスタンスであるBuilderが提供されます。これは、グラフ構造の操作に使用されます。

次に、独自のトラップを定義します。 groupedを使用して、着信要素を任意のサイズ（デフォルトでは5）のグループに結合し、その後これらのグループを送信します。 そのような各グループについて、副作用をシミュレートします：電子メール通知。

独自のトラップを定義したら、 builderインスタンスを使用してそれをグラフに追加できます。 2つの出力を持つBroadcastステップも追加します。

次に、グラフ要素間の接続を指定する必要があります。 Broadcastステップの出力の1つを電子メールトラップに接続し、グラフステップの出力を作成します。 作成したステップの入力は、 Broadcastステップの出力に直接接続されます。

注1
コンパイラは、グラフのすべての部分が正しく接続されているかどうかを判断できません。 ただし、このポイントは実行時にマテリアライザーによってチェックされるため、入力または出力にハングする要素はありません。

注2
この場合、記述したすべてのステップの形式がGraph <S、M>であることがわかります。ここで、Sは入力と出力の数とタイプを決定するフォームであり、Mは具体化された値（存在する場合）です。 ここでは、Flowフォームを扱っています。つまり、1つの入力と1つの出力があります。

最後の段階では、 liveReadingsパイプラインの追加ステップとしてnotifierを接続します。このパイプラインは次の形式を取ります。

 private final Source<Double, NotUsed> liveReadings = newFiles .via(csvPaths) .via(fileBytes) .via(csvFields) .via(readings) .via(averageReadings) .via(notifier); 

更新されたコードを実行すると、電子メール通知に関するメッセージがログにどのように表示されるかがわかります。 別の5つの値がWebソケットを通過するたびに通知が送信されます。

まとめ

この記事では、ストリーミングデータ処理の一般的な概念を検討し、Akka Streamsを使用して軽量のデータ処理パイプラインを構築する方法を学びました。 これは、Java EEで使用される従来のアプローチの代替手段です。

Akka Streamsに組み込まれたいくつかの処理ステップの使用方法、Graph DSLで独自のステップを記述する方法を検討しました。 また、Alpakkaを使用してファイルシステムとAkka HTTPプロトコルからデータをストリーミングする方法も示しました。これにより、エンドポイントにWebソケットを備えたAkka Streamsとシームレスに統合されたシンプルなWebサーバーを作成できます。

この記事のコードの完全な実例はGitHubにあります 。 異なるポイントにいくつかの追加のlogステップがあります。 コンベア内で何が起こっているかをより正確に想像するのに役立ちます。 この記事では、短くするために意図的にそれらを省略しました。