4

私の akka-streams Learn-o-thon は続きます。akka-streams アプリケーションをakka-cluster および DistributedPubSubMediator と統合したいと考えています。

パブリッシュのサポートを追加するのはかなり簡単ですが、サブスクライブの部分に問題があります。

参考までに、サブスクライバーはTypesafe サンプルで次のように指定されています。

class ChatClient(name: String) extends Actor {
  val mediator = DistributedPubSub(context.system).mediator
  mediator ! Subscribe("some topic", self)

  def receive = {
    case ChatClient.Message(from, text) =>
      ...process message...
  }
}

私の質問は、このアクターを自分のフローに統合するにはどうすればよいですか? また、ストリーム バックプレッシャーがない状態でパブリッシュ メッセージを確実に取得するにはどうすればよいですか?

あるストリームがメッセージを発行し、別のストリームがそれを消費する (購読している場合) pubsub モデルを実現しようとしています。

4

3 に答える 3

8

Actor に ActorPublisher を拡張させたいと思うでしょう。次に、そこから Source を作成し、それをストリームに統合できます。

ActorPublisher のドキュメントを参照してください: http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0.3/scala/stream-integrations.html

于 2016-02-03T03:07:01.520 に答える