java - 多个 Kafka 分区到 Akka 流

Question

嗨，我正在与 Kafka 合作和Akka Streams 。在 Kafka对于一个主题MyTestTopic我有3个分区，数据正在以高并发的方式推送到主题中，大约1000 QPS而且只会比这个更高。

下面是我的 Akka Stream Kafka Consumer 代码:

final ConsumerSettings<String, byte[]> consumerSettings =
        ConsumerSettings.create(kafkaConfig, new StringDeserializer(), new ByteArrayDeserializer())
                .withBootstrapServers("127.0.0.1:9092")
                .withGroupId("TestConsumerGroup")
                .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
                .withProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
                .withProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(timeout));

ActorMaterializer materializer = ActorMaterializer.create(system);

RestartSource.onFailuresWithBackoff(
        java.time.Duration.ofSeconds(3),
        java.time.Duration.ofSeconds(3000),
        0.2,
        () -> Consumer.atMostOnceSource(consumerSettings, Subscriptions.topics("MyTestTopic"))
                .mapAsyncUnordered(10,
                        record -> ask(rootHandler, new StreamData(record), Duration.ofSeconds(timeout))))
        .to(Sink.foreach(App::sinkParser))
        .run(materializer);

我的问题:

• 如何定义多个 Akka Stream consumers聆听不同的声音Kafka partitions因为导致 Akka Steam 的单个实例的多个分区看起来像 bottle-neck 。
• 是Akka Clustering这个问题的答案是什么？保留2 seed nodes在静态服务器上和 multiple akka stream consumers在基于云的环境中自动缩放。

我似乎无法弄清楚，我需要帮助，谢谢

Answer 1

有几种方法可以解决这个问题，具体取决于您尚未详细说明的细节:

如果您相当确定一个节点可以处理所有消息，则可以设置多个流，每个分区最多 1 个流。

对此的演变是使用 CommittablePartitionedSource这样您就可以动态创建与分区一样多的流。请注意，您需要手动提交偏移量(例如使用 Committer.sink)。

每个实例可以有一个流，并且可以部署到与分区一样多的实例；对于相同的消费者组，实例将在它们之间协调分区分配。部署多个实例时，您可能需要也可能不需要 Akka Cluster，具体取决于您要求的参与者正在执行的操作的性质。

如果每条消息的 Actor 中没有维护任何状态(请注意，这将包括 Actor 在外部数据存储上执行读取-修改-写入操作:如果您可以确保影响给定行的消息位于同一个 Kafka 中分区，您甚至可以在外部数据存储中没有 ACID)，您可能不需要 Akka 集群。

如果参与者本身是有状态的(例如，他们正在监视某些 IoT 设备)，那么您几乎肯定需要 Akka 集群、Akka 集群分片和 Akka 持久性的组合。与在外部数据存储上执行读取-修改-写入的参与者相比，始终这样做确实具有一些优势(例如，可以消除大多数读取，以有利于跟踪参与者中的状态并接受事件源)。