[翻译]akka in action之akka-stream（4 生产者与消费者之间的调解）

旧城等待， 2021-12-04 02:17 284阅读 0赞

## 生产者与消费者之间的调解 ##

下一个示例将涉及将事件广播到消费服务。到目前为止，我们已经将日志事件写入磁盘——一个含有所有事件的日志文件，以及警告、错误和关键错误的几个文件。

在日志流处理器的最终版本中, 事件将被发送到存档服务、通知服务和度量服务。

日志流处理器必须平衡供应和需求，以确保当其中一个服务应用回压时，这不会减慢日志事件的生产者。在下一节中，我们将讨论如何使用缓冲区来实现这一点。

> **集成服务**  
> 在本节的示例中, 我们希望所有服务都提供Sink。Akka-stream 有几个用于集成不提供Sources 或 Sinks的外部服务的选项。例如，mapAsync 方法带有一个Future，并会进一步向下游发出Future 的结果, 当你已经有一个使用Future的服务客户端代码时, 这可能会非常有用。  
> 与其它Reactive Streams 实现可以使用Source.fromPublisher 和 Sink.fromSubscriber 进行集成，将任何 Reactive Streams Publisher 转换为Source以及任何Subscriber 转换为Sink。也可以使用ActorPublisher 和 ActorSubscriber 特质集成Actor，在某些具体的案例中会有用。  
> 最好和最简单的选项是使用基于akka-stream库提供Source和Sink。

## 4.1 使用缓冲 ##

让我们看看处理事件的图，现在它适配于将数据发送到三个服务Sink。

![aHR0cHM6Ly9waWMxLnpoaW1nLmNvbS84MC92Mi1mMjc2NmUzNzg0MTE4MmMxYTU0ZWVjOTlmOTE1MGQyMF9oZC5qcGc][]

图1

![aHR0cHM6Ly9waWM0LnpoaW1nLmNvbS84MC92Mi02NjQ1MDY4MWU5NjRhZjI0MjY5MjgzZTVmNzI3MDcxYl9oZC5qcGc][]

图2

Broadcast添加到每个过滤流上。一个输出像先前写入到日志文件Sink，另一个用于向下游服务发送数据。（这里的期望是日志文件Sinks 非常快，所以它们没有缓冲。）MergePreferred 阶段将所有通知摘要合并到通知服务的Sink, 关键事件摘要比错误和警告摘要优先级高。关键事件摘要总是包含一个关键事件。

OK事件也使用广播进行拆分，并将它们发送到度量服务。

该图还展示了插入缓冲区的位置。缓冲区允许下游消费者在消费数据的速度上有所不同。但是当缓冲区满了，就必须做出决定。

Flow上的buffer方法需要两个参数：一个是缓冲大小和一个OverflowStrategy，这个策略决定了缓冲区即将溢出时会发生什么。OverflowStrategy 可以设置为dropHead, dropTail, dropBuffer, drop-New, backpressure, or fail之一，分别从缓冲中放弃第一个元素，缓冲中最后一个元素，整个缓冲，或者最新的元素了，或者当缓冲满的时候应用背压；或者让整个流失败。选择哪一个有赖于应用的需求和具体的案例什么最重要。

在这个例子中，已经做出决定，即使在高负载下，所有事件都必须存档，这意味着当不能将事件写入归档服务Sink时，日志流处理器流应该失败。生产者可以稍后再试。设置一个大的缓冲区，如果归档Sink在一段时间内缓慢响应，则进行缓冲。

下面的代码展示了如何在图中设置缓冲。

![aHR0cHM6Ly9waWMyLnpoaW1nLmNvbS84MC92Mi0xZmRhMzg4MTM4YWI2OTQyYjNmN2QzNDYyMmRmYjViNV9oZC5qcGc][]

缓冲

如果通知服务Sink接收慢，则在高负载下可以丢弃最旧的警告。不能删除错误摘要。关键错误没有缓冲，所以默认情况下流会使用背压。

val bcast = builder.add(Broadcast[Event](5))
    val wbcast = builder.add(Broadcast[Event](2))
    val ebcast = builder.add(Broadcast[Event](2))
    val cbcast = builder.add(Broadcast[Event](2))
    val okcast = builder.add(Broadcast[Event](2))
    
    val mergeNotify = builder.add(MergePreferred[Summary](2))
    val archive = builder.add(jsFlow)

MergePreferred 始终有一个首选端口和多个次要端口, 在本例中为2个。

首先, 下面的代码展示了所有图形节点是如何连接的。

![aHR0cHM6Ly9waWM0LnpoaW1nLmNvbS84MC92Mi0yOTFkN2NhYzlkMWNhNWI2YmM4NDQ2ZGU5N2E1MTdmM19oZC5qcGc][]

连接图形节点

在下一节中，我们将研究如何在流中以不同的速率处理元素。将使用一种特殊的流操作来分离流的一方与另一方的速率。

[aHR0cHM6Ly9waWMxLnpoaW1nLmNvbS84MC92Mi1mMjc2NmUzNzg0MTE4MmMxYTU0ZWVjOTlmOTE1MGQyMF9oZC5qcGc]: /images/20211203/ff4bd00530cb4da1bc3714097f7d4dbe.png
[aHR0cHM6Ly9waWM0LnpoaW1nLmNvbS84MC92Mi02NjQ1MDY4MWU5NjRhZjI0MjY5MjgzZTVmNzI3MDcxYl9oZC5qcGc]: /images/20211203/5fa3a1f833044b919b2fc66df6ee01a7.png
[aHR0cHM6Ly9waWMyLnpoaW1nLmNvbS84MC92Mi0xZmRhMzg4MTM4YWI2OTQyYjNmN2QzNDYyMmRmYjViNV9oZC5qcGc]: /images/20211203/571de6603fea46d49efcf8526696d422.png
[aHR0cHM6Ly9waWM0LnpoaW1nLmNvbS84MC92Mi0yOTFkN2NhYzlkMWNhNWI2YmM4NDQ2ZGU5N2E1MTdmM19oZC5qcGc]: /images/20211203/c3fbe17d5418469fa9843f2b97e3a7fe.png