处理消息后提交Kafka消费者补偿的好模式是什么？

Question

我正在使用Akka Streams 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来使其更清洁和更易于更改。它可以让你产生一个支持你的消息部分的分支，而另一个分支可以执行所有需要的业务逻辑。

图的一个例子可以（忽略所有的更多，更清晰的样板）：

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

下面是一个使用@斯特凡诺 - 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) 
    }