リアクティブストリームの紹介-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



ステップがあります。 コンベア内で何が起こっているかをより正確に想像するのに役立ちます。 この記事では、短くするために意図的にそれらを省略しました。