多个Kafka分区到Akka流

2019年11月26日 21点热度 0条评论

嗨,我正在使用KafkaAkka Streams。在主题KafkaMyTestTopic中,我有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保留在自动扩展上。

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

解决方案如下:

有两种方法可以解决此问题,具体取决于您尚未详细说明的细节:

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

对此的改进将是使用CommittablePartitionedSource,以便您动态创建与分区一样多的流。请注意,您需要手动提交偏移量(例如,使用Committer.sink)。

每个实例可以有一个流,最多可以部署与分区一样多的实例。对于相同的使用者组,实例将在它们之间协调分区分配。部署多个实例时,您可能需要Akka群集,也可能不需要,这取决于您正在执行 操作的角色的性质。

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

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