Lagom服务消耗来自Kafka的输入

Question

我试图弄清楚如何使用Lagom来消费来自通过Kafka进行通信的外部系统的数据。

我遇到了这个section of Lagom documentation，它描述了Lagom服务如何通过订阅它的主题与另一个Lagom服务进行通信。

helloService
  .greetingsTopic()
  .subscribe // <-- you get back a Subscriber instance
  .atLeastOnce(
  Flow.fromFunction(doSomethingWithTheMessage)
)

但是，当您想要订阅包含某个随机外部系统生成的事件的Kafka主题时，适当的配置是什么？

这个功能需要某种适配器吗？为了澄清，我现在有这个：

object Aggregator {
  val TOPIC_NAME = "my-aggregation"
}

trait Aggregator extends Service {
  def aggregate(correlationId: String): ServiceCall[Data, Done]

  def aggregationTopic(): Topic[DataRecorded]

  override final def descriptor: Descriptor = {
    import Service._

    named("aggregator")
      .withCalls(
        pathCall("/api/aggregate/:correlationId", aggregate _)
      )
      .withTopics(
        topic(Aggregator.TOPIC_NAME, aggregationTopic())
          .addProperty(
            KafkaProperties.partitionKeyStrategy,
            PartitionKeyStrategy[DataRecorded](_.sessionData.correlationId)
          )
      )
      .withAutoAcl(true)
  }
}

我可以通过简单的POST请求调用它。但是，我希望通过使用来自某些（外部）Kafka主题的Data消息来调用它。

我想知道是否有这种方式以类似于这个模型的方式配置描述符：

override final def descriptor: Descriptor = {
  ...
  kafkaTopic("my-input-topic")
    .subscribe(serviceCall(aggregate _)
    .withAtMostOnceDelivery
}

我遇到了这个discussion on Google Groups，但是在OPs的问题中，我没有看到他实际上正在做EventMessage来自some-topic的任何事情，除了将他们引导到他的服务定义的主题。

编辑＃1：进度更新

看一下文档，我决定尝试以下方法。我又添加了2个模块，aggregator-kafka-proxy-api和aggregator-kafka-proxy-impl。

在新的api模块中，我定义了一个没有方法的新服务，但是一个主题代表了我的Kafka主题：

object DataKafkaPublisher {
  val TOPIC_NAME = "data-in"
}

trait DataKafkaPublisher extends Service {
  def dataInTopic: Topic[DataPublished]

  override final def descriptor: Descriptor = {
    import Service._
    import DataKafkaPublisher._

    named("data-kafka-in")
      .withTopics(
        topic(TOPIC_NAME, dataInTopic)
          .addProperty(
            KafkaProperties.partitionKeyStrategy,
            PartitionKeyStrategy[SessionDataPublished](_.data.correlationId)
          )
      )
      .withAutoAcl(true)
  }
}

在impl模块中，我只是做了标准实现

class DataKafkaPublisherImpl(persistentEntityRegistry: PersistentEntityRegistry) extends DataKafkaPublisher {
  override def dataInTopic: Topic[api.DataPublished] =
    TopicProducer.singleStreamWithOffset {
      fromOffset =>
        persistentEntityRegistry.eventStream(KafkaDataEvent.Tag, fromOffset)
          .map(ev => (convertEvent(ev), ev.offset))
    }

  private def convertEvent(evt: EventStreamElement[KafkaDataEvent]): api.DataPublished = {
    evt.event match {
      case DataPublished(data) => api.DataPublished(data)
    }
  }
}

现在，为了实际使用这些事件，在我的aggregator-impl模块中，我添加了一个“订阅者”服务，它接受这些事件，并在实体上调用适当的命令。

class DataKafkaSubscriber(persistentEntityRegistry: PersistentEntityRegistry, kafkaPublisher: DataKafkaPublisher) {

  kafkaPublisher.dataInTopic.subscribe.atLeastOnce(
    Flow[DataPublished].mapAsync(1) { sd =>
      sessionRef(sd.data.correlationId).ask(RecordData(sd.data))
    }
  )

  private def sessionRef(correlationId: String) =
    persistentEntityRegistry.refFor[Entity](correlationId)
}

这有效地允许我在Kafka主题“data-in”上发布消息，然后将其代理并转换为RecordData命令，然后发布给要使用的实体。

但是，对我来说，这似乎有些苛刻。我与Lagom internals联系到Kafka。我无法轻易交换数据来源。例如，如果我愿意，我将如何使用来自RabbitMQ的外部消息？如果我想从另一个Kafka消费（不同于Lagom使用的那个）怎么办？

编辑＃2：更多文档

我在Lagom文档上发现了一些文章，特别是这个：

Consuming Topics from 3rd parties

您可能希望您的Lagom服务使用在Lagom中未实现的服务上生成的数据。在这种情况下，如“服务客户端”部分所述，您可以在Lagom项目中创建第三方服务API模块。该模块将包含一个服务描述符，声明您将使用的主题。一旦实现了ThirdPartyService接口和相关类，就应该添加第三方服务api作为对fancy-service-impl的依赖。最后，您可以使用ThirdPartyService中描述的主题，如订阅主题部分中所述。

Answer 1

我不使用lagom所以这可能只是一个想法。但是因为akka-streams是lagom的一部分（至少我认为是这样） - 从这个解决方案到你需要的东西应该很容易。

我使用akka-stream-kafka，这非常好（我只做了一个原型）

在使用消息时，您可以执行以下操作：

     Consumer
      .committableSource(
          consumerSettings(..), // config of Kafka
          Subscriptions.topics("kafkaWsPathMsgTopic")) // Topic to subscribe
      .mapAsync(10) { msg =>
        business(msg.record) // do something
      }

检查写得好的documentation

我在这里找到的完整例子：PathMsgConsumer

Answer 2

Alan Klikic在Lightbend讨论论坛here上提供了答案。

第1部分：

如果您只在业务服务中使用外部Kafka群集，那么您可以仅使用Lagom Broker API实现此功能。所以你需要：

1. 使用仅包含主题定义的服务描述符创建API（此API未实现）
2. 在您的业务服务中根据您的部署配置kafka_native（正如我在上一篇文章中提到的那样）
3. 在您的业务服务中，从＃1中创建的API注入服务，并使用Lagom Broker API订阅者订阅它

在Lagom Broker API订户中的抵消提交是开箱即用的。

第2部分：

Kafka和AMQP消费者实施需要持久的akka​​流。所以你需要处理断开连接。这些可以通过两种方式完成：

1. 通过将其包裹在演员中来控制peristant akka流。您初始化您在actor preStart和流管道流上的流完成到将停止它的actor。如果流完成或失败，则actor将停止。然后使用重新启动策略将actor包含在actor backoff中，这将在完成或失败的情况下重新启动actor并重新初始化Flow
2. akka streams延迟重启与退避阶段

就个人而言，我使用＃1并没有尝试＃2。

初始化＃1的倒流actor或＃2的流可以在您的Lagom组件特征中完成（基本上在您使用Lagom Broker API进行订阅的同一位置）。

确保在配置使用者时设置使用者组以确保避免重复使用。您可以像Lagom一样使用描述符中的服务名称作为使用者组名称。