0

我需要从拥有数百万数据的 Kafka 主题中消费。从主题中阅读后，我需要对其进行转换并将其写入另一个主题。我能够使用来自主题的消息，通过多个线程处理数据并写入另一个主题。我按照这里的例子https://projectreactor.io/docs/kafka/1.3.5-SNAPSHOT/reference/index.html#concurrent-ordered

这是我的代码：

public Flux<?> flux() {
            KafkaSender<Integer, Person> sender = sender(senderOptions());
            return KafkaReceiver.create(receiverOptions(Collections.singleton(sourceTopic)))
                                .receive()
                                .map(m -> SenderRecord.create(transform(m.value()), m.receiverOffset()))
                                .as(sender::send)
                                .doOnNext(m -> m.correlationMetadata().acknowledge())
                                .doOnCancel(() -> close());
        }

由于数据量大，我有多个消费者可供阅读，并且正在考虑添加不同的阅读器线程以从主题中读取。然而，reactor-kafka文档提到 KafkaReceiver 不是线程安全的，因为底层的 KafkaConsumer 不能被多个线程同时访问。

我正在寻找有关同时阅读某个主题的建议。

4

1 回答 1

1

所以基本上你正在寻找所谓的消费者组，你可以运行的最大并行消费受到你的主题的分区数量的限制。

Kafka 消费者组机制允许您将一个主题的消费工作分离给属于同一组的不同“读者”，工作将被划分为组中的每个消费者将单独负责一个分区（1 个或多个，基于组中的消费者数量和主题的分区数量）

于 2021-07-27T22:52:05.947 回答