如何创建具有背压和控制的Akka流

Question

我需要创建一个具有以下接口的函数：

import akka.kafka.scaladsl.Consumer.Control

object ItemConversionFlow {

def build(config: StreamConfig): Flow[Item, OtherItem, Control] = {
    // Implementation goes here
}

我的问题是，我不知道如何定义符合上述接口的流。

当我做这种事的时候

val flow = Flow[Item]
    .map(item => doConversion(item)
    .filter(_.isDefined)
    .map(_.get)

结果类型为Flow[Item，OtherItem，NotUsed]。到目前为止，我还没有在Akka文档中找到任何东西。还有akka上的功能。流动scaladsl。流只提供“未使用”而不是控制。如果有人能给我指出正确的方向那就太好了。

一些背景：我需要设置几个只在转换部分区分的管道。这些管道是主流的子流，可能由于某种原因而停止（相应的消息到达某个Kafka主题）。因此我需要控制部分。其想法是创建一个图形模板，我只是在其中插入所提到的流作为参数（一个返回它的工厂）。对于特定情况，我们有一个有效的解决方案。一般来说，我需要这种流动。

Answer 1

实际上你有背压。然而，想想你真正需要的背压是什么。。。您没有使用异步阶段来增加吞吐量。。。例如背压可以避免快速生产商过度增长的用户https://doc.akka.io/docs/akka/2.5/stream/stream-rate.html.在您的示例中，您的流将根据doConversion完成所需的时间向发布者请求新元素。

如果您想获取流的结果，请使用toMat或viaMat。例如，如果您的流发出Item并将它们转换为其他项目：

val str = Source.fromIterator(() => List(Item(Some(1))).toIterator)
  .map(item => doConversion(item))
  .filter(_.isDefined)
  .map(_.get)
  .toMat(Sink.fold(List[OtherItem]())((a, b) => {
      // Examine the result of your stream
      b :: a
    }))(Keep.right)
  .run()

str将是未来[列表[其他项目]]。尝试将其推断到您的情况。

或者将toMat与KillSwitches一起使用，“创建[[FlowShape]]的新[[Graph]]，该[[FlowShape]]具体化到外部交换机，允许外部完成该唯一具体化*。不同的具体化会导致不同的独立开关。”

  def build(config: StreamConfig): Flow[Item, OtherItem, UniqueKillSwitch] = {
    Flow[Item]
      .map(item => doConversion(item))
      .filter(_.isDefined)
      .map(_.get)
      .viaMat(KillSwitches.single)(Keep.right)
  }


  val stream = 
    Source.fromIterator(() => List(Item(Some(1))).toIterator)
    .viaMat(build(StreamConfig(1)))(Keep.right)
    .toMat(Sink.ignore)(Keep.both).run

  // This stops the stream
  stream._1.shutdown()

  // When it finishes
  stream._2 onComplete(_ => println("Done"))