使用akka streams kafka处理流终止

iqjalb3h 于 2021-06-07 发布在 Kafka

关注(0)|答案(1)|浏览(339)

我正在使用akkastreamsKafka库与Kafka经纪人进行交互。我有以下具有无限缓冲区的流定义：

def producerStream[T: MessageType](producerProperties: Map[String, String]) = {
    val streamSource = source[T](producerProperties)
    val streamFlow = flow[T](producerProperties)
    val streamSink = sink(producerProperties)
    streamSource.via(streamFlow).to(streamSink).run()
  }

我从actor示例中调用producerstream。我有一个类型的消息：

case class InitiateStream[T](publisher: ActorRef, anyMessage: T)

在我的actor中，我有以下接收块：

override def receive: Receive = super.receive orElse {
    case StartProducerStream(publisherActor, DefaultMessage) =>
      producerStream[DefaultMessage](cfg.producerProps)
      // context.become(active)

    case other => println(s"SHIT !! Got unknown message: $other")
  }

发布者最终将获得应该推送到相应Kafka主题的消息。
现在我的问题是，我是否需要费心关闭producerstream？
我正在考虑在启动流时立即执行context.been（active（producerstream）），在active方法中，我将根据destroysteam消息处理流终止。需要这个吗？你们怎么看？

scala akka apache-kafka akka-stream

来源：https://stackoverflow.com/questions/48169177/handling-streams-termination-with-akka-streams-kafka

1条答案

按热度按时间

wooyq4lh1#

如果使用akka kafka使用者，则源将无限期地发出事件，直到主题中的事件可用为止。为什么要关闭生产商？
我´我建议您拨打以下电话开始您的来源：

def producerSource[T: MessageType](producerProperties: Map[String, String]) = {
    val streamSource = source[T](producerProperties)
    val streamFlow = flow[T](producerProperties)
    val streamSink = sink(producerProperties)
    streamSource.via(streamFlow)
  }

def startStream[A](source: Source[A, NotUsed])(
  implicit am: ActorMaterializer): (UniqueKillSwitch, Future[Done]) = {
   source
   .viaMat(KillSwitches.single)(Keep.right)
   .toMat(Sink.ignore)(Keep.both)
   .run
}

var streamHandler : Option[UniqueKillSwitch] = None

  override def receive: Receive = super.receive orElse {
    case StartProducerStream(publisherActor, DefaultMessage) =>
      streamHandler = Some(producerStream[DefaultMessage](cfg.producerProps))
      context.become(active)

    case other => println(s"SHIT !! Got unknown message: $other")
  }

对于主动行为：

val active: PartialFunction[Any, Unit] = 
     case DestroyStream => 
        if(streamHandler.isDefined)
            streamHandler.shutdown

赞(0）回复(0）举报 2021-06-07

我来回答

使用akka streams kafka处理流终止

1条答案

相关问题

热门标签

最新问答