Streams」とは現在のWebでは膨大な量のデータをリアルタイムでやりとりしています。いったデータは普通にダウンロード/アップロードしていては時間がかかり過ぎてしまうので、ストリーム処理(無限に発生するデータを継続的に処理する仕組み)が必要になります。下記などはストリームが妥当な処理方法です。テラバイト級ファイルの処理(メモリに展開し切れない)通信処理(処理が継続的に発生し続ける)Stream処理では、受信側(Subscriber)が処理しきれないデータを送信し続けるとバッファーがあふれてしまうため、送信側(Publisher)がデータを送り過ぎないようにする必要があります。Streams」(ノンブロッキングで非同期なストリーム処理の仕様)の基本方針です。Streamsなのです。Streamsは処理する要素の入力と出力がはっきり決まっているため、図を作成するように分かりやすく処理フローを記述できます。処理については副作用がなく、再利用も可能になっています。Streamsを使うために覚えておきたいキーワードについて解説します。Sourceデータの源泉であり、出力を行う役割を持ちます(Publisher)。そのため、入力用チャンネルは持たず、出力チャンネルを1つだけ持っています。ourceSinkデータの出力先です。1つの入力用チャンネルを持ち、出力用チャンネルは持っていません。SinkFlow入力チャンネルと出力チャンネルを1つずつ持っており、SourceとSinkの間で、データ処理を行えます。SourceとFlowと接続すれば新たなSourceに、SinkとFlowを接続すれば新たなSinkになります。StreamsでStream処理を実行する環境の抽象モデルです。「ActorMaterializer」クラスを使うと、アクターでの実行が可能になります。RunnableGraphSource、Flow、Sinkを接続した結果が図4の「RunnableGraph」です。RunnableGraphとは実行可能な状態のGraphです。このRunnableGraphにrun関数を実行することで、Source、Flow、Sinkの処理が評価されます。Streamsを使ってみましょう。必要なクラスをインポートし、ActorMaterializerのインスタンスを作成します。ActorMaterializer()次にSourceを作ります。Sourceは下記のように、幾つかの方法で作れます。Streams"))コラム「Future型とは」Futureは、まだ存在しない処理結果を抽象化した型です。通常、Futureは並行処理され後で取り出すことが可能になっています。このため、Futureを使用した処理は非同期でノンブロッキングになることが多くあります。次にFlowを作ります。ここでは「Sourceから要素を受け取り、処理(値を2倍)した後Sinkへ渡す」処理を記述します。2)最後に、Sinkを作ります。ここでは受け取った要素全てに対して標準出力(println)を行います。Sink.foreach[Int](println)Source→FLow→Sinkをつないで、RunnableGraphとして実行してみます。下記プログラムではSourceで1〜5のRangeオブジェクトを作成し、Flowでその値を2倍にし、Sinkで出力しています。graph.run()}SourceとFlowはvia関数で接続し、Sinkはto関数で接続します。toはRunnableGraphを返すので、それに対してrunを実行することで実際の処理を行えます。Streamsを使ったHTTPサーバのサンプルもう少し具体的なサンプルプログラムを作ってみましょう。Streamsを使ってファイルのコピーを行います。入力ストリームから受け取ったデータを全て出力ストリームへ書き込むだけの単純なサンプルです。Source作成はPaths.getを使用してパスを取得し、akka.stream.scaladsl.FileIOクラスのfromPathを使用して作成します。FileIO.fromPath(inPath)同じようにSinkも作ります。FileIO.toPath(outPath,Set(CREATE,WRITE,APPEND))SourceとSinkは、どちらもストリームのエンドポイントです。今回はSourceとSinkを接続し、RunnableGraphを生成します。source.to(sink)これでSourceからデータを受け取ってSinkへ送るRunnableGraphができましたが、これだけではまだコピーは実行されません。run関数を実行することでファイルのコピーが実行されます。Streamsを使ったHTTPサーバのサンプル全体のソースコードは下記です。Streamsについて、実際にサンプルを動かしてみました。Streamsはリアクティブアプリを構築する上で重要です。Streamsを含んでいるため、依存関係を追加しなくても使えます。うた)クラスメソッド勤務の新しもの好きプログラマーです。数年前に東京から山口県に引っ越し、現在もリモート勤務しています。最近の趣味は空手とぬか漬け作り。8/9はマイクロサービス、リアクティブに向かう——MVCは生き残れるのかデジタルトランスフォーメーション時代に生き残れるエンジニアに求められるものとは何か。日本のJavaコミュニティで存在感を示し続け、現在は日本マイクロソフトでJavaエバンジェリストとして活動する寺田佳央氏に聞いた。EJB、SOA、マイクロサービスへと至る大規模システム向けアーキテクチャの変遷2000年前後からのアプリケーションアーキテクチャやEJB、SOAに触れながら、今後、大規模システム構築で主流になるであろう「マイクロサービス」アーキテクチャの意義と価値を考える。メールマガジンをご購読ください。申し込みページへITmediaはアイティメディア株式会社の登録商標です。