从 EventStream 中创建源答案

【问题标题】：Creating a Source out of an EventStream从 EventStream 中创建源
【发布时间】：2016-05-19 14:33:37
【问题描述】：

我正在使用 PlayFramework 2.5.3 并希望从 akka.event.EventStream 创建一个 akka.stream.scaladsl.Source（事件流是演员系统的一部分）。事件流会产生某种类型的事件，因此我需要订阅该特定类型的事件并使用play.api.mvc.Results.chunked 推送它们。有没有什么简单的方法可以使用 Akka Streams 2.4.5 创建这样的Source？

【问题讨论】：

标签： scala playframework server-sent-events akka-stream reactive-streams

【解决方案1】：

您可以将Source.actorRef 与订阅一起使用。 Source.actorRef 是一个具体化为 ActorRef 的源，因此您可以这样做：

// choose the buffer size of the actor source and how the actor
// will react to its overflow
val eventListenerSource = Source.actorRef[YourEventType](32, OverflowStrategy.dropHead)

// run the stream and obtain all materialized values
val (eventListener, ...) = eventListenerSource
    .viaMat(...)(Keep.left)
    <...>
    .run()

// subscribe the source actor to the stream
actorSystem.eventStream.subscribe(eventListener, classOf[YourEventType])

// now events emitted by the source will go to the actor
// and through it to the stream

注意actorRef source 有一些限制，比如它的内部缓冲区自然不支持背压溢出策略。您可能希望将Source.actorPublisher 与扩展ActorPublisher[YourEventType] 特征的演员一起使用，它会给您更多的控制权。但是，由于EventStream 是一个纯粹的基于推送的源，因此使用ActorPublisher 将无法比使用Source.actorRef 做更多的事情，因此您不妨使用更简单的方法。

【讨论】：

您能否解释一下为什么必须先调用run()（即实现流）才能获得对actor引用的引用？
@Mihai238 这是因为actor引用是Source.actorRef的物化值。物化值，顾名思义，是在流物化时产生的值。因为Source.actorRef 是可以多次物化的流的蓝图，所以它必须为每个物化提供单独的ActorRef。毕竟，如果它只在所有实现中提供一个ActorRef，它就不会很有用。