Akka流的多个Kafka分区

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群集，也可能不需要，这取决于您正在ask的角色正在做什么的性质。

如果每个消息的actor都没有保持任何状态（请注意，这将包括actor在外部数据存储上执行读-修改-写操作：如果您可以确保影响给定行的消息位于同一Kafka分区中，您甚至可以在该外部数据存储中不使用ACID的情况下使用），您可能不需要Akka群集。

[如果参与者本身是有状态的（例如，他们正在遮盖某些物联网设备），那么您几乎可以肯定希望将Akka Cluster，Akka Cluster Sharding和Akka Persistence组合在一起。相对于参与者在外部数据存储上执行读-修改-写操作，这种方式确实具有一些优势（例如，可以消除大多数读取，而有利于跟踪参与者中的状态并采用事件源）。