akka-stream

0熱

1答えて

Playframework Java - コントローラでチャンクされたHTTPリクエストを受け取る方法

私はJavaでプレイコントローラを持っています。リクエストは15MBのサイズです。このリクエストは、APIGEEからストリーム方式で送信されます。私は、コントローラー内の生ボディパーサーを使用してデータをファイルに書き込みました。しかし、私はそれが一度にファイルに要求データ全体を書き込むと思います。つまり、リクエスト全体が1つの時点から1つの時点でメモリに格納されてからファイルに書き込まれる必要が

0熱

2答えて

フローがエラーで失敗しましたリアクティブストリーム仕様に違反してシャットダウン

Akka Streamsを使用しているときにエラー処理が正しく行われないようです。だからこれは私のコード var db = Database.forConfig("oracle") var mysqlDb = Database.forConfig("mysql_read") var mysqlDbWrite = Database.forConfig("mysql_write") impl

4熱

1答えて

akka-stream + akka-httpライフサイクル

TLDR：リクエストごとにストリームを具体化する（つまり、短命のストリームを使用する）方が良いですか？ストリームの？詳細：私はHTTPリクエストを受け取り、いくつかのサードパーティのダウンストリームサービス（私によって管理されていない）に分散し、結果を集約して戻します。私はクライアントの実装にakka-httpを使用しており、サーバー用にスプレーしています（レガシー、時間の経過とともにakka-

1熱

1答えて

タイムアウト後にコンシューマがWakeupExceptionで中断されました。メッセージ：null。 akka.kafka.consumer.wakeup-timeoutの現在の値は3000ミリ秒です

akkaストリームを使用してkafkaからのメッセージの単純なコンシューマを書き込もうとしています。 build.sbt "com.typesafe.akka" %% "akka-stream-kafka" % "0.17" 私のコード object AkkaStreamskafka extends App { // producer settings implicit

1熱

1答えて

Akkaストリーム - データを失うことなくグループ化したシャットダウンストリーム

バッチリクエストを作成する要素とシンクを持つソースがあります。私はKillSwitchを使用して、任意の時点でグラフをシャットダウンできます。が val source = Source.tick(10.millis, 10.millis, "tick").grouped(500) val (switch, _) = source.viaMat(KillSwitches.single)(K

0熱

1答えて

ファンアウトアウトカラーストリーム処理を同時に行う方法

Akka Streamsを使用して、一連の依存ストリームを同時に処理しようとしています。このような何か： val concurrency = 2 Source( (1 to 5).toStream.map(i => { println(s"1: Emitting $i") i.toString })) .mapAsyncUnordered(con

2熱

1答えて

akka-httpでHttpResponseに未来[ソース[...]]をマーシャリングする方法は？

IはFuture[Source[List[String]]]（CSVファイルの最初の二行）を返す以下の方法た：アレイのJSON配列にそれをマーシャリング def get(url: String, charset: String, delimiter: Char, quote: Char, escape: Char) = { val scanner = CsvParsing.lineSca

0熱

1答えて

アッカカフカストリームsupervison戦略は、私がアッカストリームカフカのアプリケーションを実行していると私は、そのようなブローカーがダウンして、ストリームの消費者は、停止タイムアウト後に死亡した場合、そのストリームの消費者の監督の戦略を組み込みたい

を動作していませんスーパーバイザはコンシューマを再起動できます。 UserEventStream： import akka.actor.{Actor, PoisonPill, Props} import akka.kafka.{ConsumerSettings, Subscriptions} import akka.kafka.scaladsl.Consumer import akka.st

0熱

1答えて

アッカHTTPは、それは完璧に動作し、次のメッセージが表示さ私はポート8080上でローカルに実行されているシンプルなアッカHTTPサーバを持っているドッキングウィンドウコンテナ

に途中でシャットダウンします：私はの.jarを作成するためにsbt-assemblyを使用しています Started server at 127.0.0.1:8080, press enter to kill server をファイル。、それを docker run {my namespace}/my-app を今すぐ： FROM openjdk:8u131 ADD target/sc

0熱

1答えて

スパークストリーミングのakkaストリーム：メッセージはアクターに配信されません。死んだ手紙を受け取る

私はakkaストリーミングを始めました。私はgithubから以下の例を実行しています。しかし、 "Helloer"アクターへのメッセージは、出力コンソールで受信されず、表示されません。 StreamingApp.scala import _root_.akka.actor.{ Actor, Props } import org.apache.spark._ import org.apache.