Akka-Http Websockets：如何向消费者发送相同的数据流答案

【问题标题】：Akka-Http Websockets: How to Send consumers the same stream of dataAkka-Http Websockets：如何向消费者发送相同的数据流
【发布时间】：2016-02-08 18:03:26
【问题描述】：

我有一个客户端可以连接的 WebSocket 我也有一个使用 akka-streams 的数据流。我怎样才能让所有客户都获得相同的数据。目前他们似乎在争夺数据。

谢谢

【问题讨论】：

尝试实现或使用像 Redis 这样的发布/订阅代理通道系统...并记住您可以仅使用 Flow 或使用 Flow 和 Sink 来处理 websocket。 akka 数据流可以通过将数据发布到通道的接收器来完成。
@salc2 我认为我需要的是一个事件流，它允许点对多的通信解决方案。

标签： scala akka akka-stream akka-http

【解决方案1】：

您可以做的一种方法是让一个演员扩展 ActorPublisher 并让它订阅到一些消息。

class MyPublisher extends ActorPublisher[MyData]{

  override def preStart = {
    context.system.eventStream.subscribe(self, classOf[MyData])
  }

  override def receive: Receive = {

    case msg: MyData ⇒
      if (isActive && totalDemand > 0) {
        // Pushes the message onto the stream
        onNext(msg)
      }
  }
}

object MyPublisher {
  def props(implicit ctx: ExecutionContext): Props = Props(new MyPublisher())
}

case class MyData(data:String)

然后您可以将该演员用作流的源：

val dataSource = Source.actorPublisher[MyData](MyPublisher.props(someExcutionContext))

然后您可以从该数据源创建一个流并应用转换将数据转换为 websocket 消息

val myFlow = Flow.fromSinkAndSource(Sink.ignore, dataSource map {d => TextMessage.Strict(d.data)})

然后您可以在路线处理中使用该流程。

path("readings") {
  handleWebsocketMessages(myFlow)
}

通过对原始流的处理，您可以将数据发布到事件流，并且该参与者的任何实例都会将其拾取并放入提供 Websocket 服务的流中。

  val actorSystem = ActorSystem("foo")

  val otherSource = Source.fromIterator(()  => List(MyData("a"), MyData("b")).iterator)

  otherSource.runForeach { msg ⇒ actorSystem.eventStream.publish(MyData("data"))}

然后，每个套接字将拥有自己的参与者实例，以向其提供来自单一来源的数据。

【讨论】：

谢谢艾尔。我试图在 cmets 中避开上述类型的解决方案。这似乎是一个非常好的解决问题的方法。我目前正在尝试实现这一点。谢谢你的例子。