处理消息后，用于承诺KAFKA消费者偏移的好模式是什么?

Question

我正在使用 akka流kafka 将kafka消息传输到远程服务。我要确保服务准确地接收到每条消息一次（最环境和最高点交付）。

。

这是我想到的代码：

  private def startFlow[T](implicit system: ActorSystem, config: Config, subscriber: ActorRef,
                           topicPattern: String,
                           mapCommittableMessageToSinkMessage: Function[CommittableMessage[String, String], T]) {
    val groupId = config.getString("group-id")
    implicit val materializer = ActorMaterializer()
    val consumerSettings = ConsumerSettings(system, new StringDeserializer, new StringDeserializer)
      .withGroupId(groupId)
      .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
    implicit val timeout = Timeout(5 seconds) // timeout for reply message on ask call below
    import system.dispatcher // the ExecutionContext that will be used in ask call below
    Consumer.committableSource(consumerSettings, Subscriptions
      .topicPattern(topicPattern))
      .map(message => (message, mapCommittableMessageToSinkMessage(message)))
      .mapAsync(1)(tuple => ask(subscriber, tuple._2).map(_ => tuple._1))
      .mapAsync(1)(message => message.committableOffset.commitScaladsl())
      .runWith(Sink.ignore)
  }

如代码所示，它映射了原始消息的元组，以及传递给订阅者的转换消息（发送给远程服务的演员）。元组的目的是在订户完成处理后提交偏移。

关于它的某些东西似乎是一个反模式，但我不确定更好的方法。有更好的建议吗？

谢谢！

Answer 1

使用GraphDSL可以保持更清洁，更易于更改的方法。它可以使您在邮件的Committable部分上产生图形的分支，而另一个分支可以执行所有必要的业务逻辑。

可以是图形的一个示例（省略所有样板以获得更好的清晰度）：

val src = Consumer.committableSource(consumerSettings, Subscriptions
      .topicPattern(topicPattern))
val businessLogic = Flow[CommittableMessage[String, String]].mapAsync(1)(message => ask(subscriber, mapCommittableMessageToSinkMessage(message)))
val snk = Flow[CommittableMessage[String, String]].mapAsync(1)(message => message.committableOffset.commitScaladsl())
      .runWith(Sink.ignore)  // look into Sink.foldAsync for a more compact re-write of this part
src ~> broadcast
       broadcast ~> businessLogic ~> zip.in0
       broadcast         ~>          zip.in1
                                     zip.out.map(_._2) ~> snk

Answer 2

这是使用 @stefano-bonetti方法在上面的答案中使用的完整代码：

  private def startStream[T](implicit system: ActorSystem, config: Config, subscriber: ActorRef,
                             topicSuffix: String,
                             convertCommittableMessageToSubscriberMessage: Function[CommittableMessage[String, String], T]) {
    val groupId = config.getString("group-id")
    val subscriberName = subscriber.path.name
    val customerId = config.getString("customer-id")
    val topicPattern = s"^$customerId\.$topicSuffix$$"
    implicit val materializer = ActorMaterializer()
    val consumerSettings = ConsumerSettings(system, new StringDeserializer, new StringDeserializer)
      .withGroupId(s"$groupId.$subscriberName")
      .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
    implicit val timeout = Timeout(5 seconds) // timeout for reply message on ask call below
    import system.dispatcher // the ExecutionContext that will be used in ask call below
    val src = Consumer.committableSource(consumerSettings, Subscriptions.topicPattern(topicPattern))
    val businessLogic = Flow[CommittableMessage[String, String]]
      .mapAsync(1)(message => subscriber.ask(convertCommittableMessageToSubscriberMessage(message)))
    val snk = Flow[CommittableMessage[String, String]]
      .mapAsync(1)(message => message.committableOffset.commitScaladsl())
      .to(Sink.ignore)
    val decider: Supervision.Decider = {
      case e => {
        system.log.error("error in stream", e)
        Supervision.Stop
      }
    }
    val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
      import GraphDSL.Implicits._
      val broadcast = builder.add(Broadcast[CommittableMessage[String, String]](2))
      val zip = builder.add(Zip[Any, CommittableMessage[String, String]])
      src ~> broadcast
      broadcast ~> businessLogic ~> zip.in0
      broadcast ~> zip.in1
      zip.out.map(_._2) ~> snk
      ClosedShape
    })
      .withAttributes(ActorAttributes.supervisionStrategy(decider))
      .run(materializer)
  }