我如何同时处理ReactorKafka流的主题和分区与自动确认？

Question

我试图实现Kafka主题分区的并发处理使用反应堆Kafka与自动确认。这里的文档使它看起来像是可能的：

http://projectreactor.io/docs/kafka/militare/reference/#并发订购

这与我尝试的唯一不同之处在于我使用的是自动确认。

public class KafkaFluxFactory<K, V> {

    private final Map<String, Object> properties;

    public KafkaFluxFactory(Map<String, Object> properties) {
        this.properties = properties;
    }

    public Flux<ConsumerRecord<K, V>> receiveAuto(Collection<String> topics, Scheduler scheduler) {
        return KafkaReceiver.create(ReceiverOptions.create(properties).subscription(topics))
            .receiveAutoAck()
            .flatMap(flux -> flux.groupBy(this::extractTopicPartition))
            .flatMap(topicPartitionFlux -> topicPartitionFlux.publishOn(scheduler));
    }

    private TopicPartition extractTopicPartition(ConsumerRecord<K, V> record) {
        return new TopicPartition(record.topic(), record.partition());
    }
}

对我做错了什么有什么想法吗？

Answer 1

经过多次反复试验，再加上重新考虑我想要完成的任务，我意识到我试图在一段代码中解决两个问题。

我需要的两件事是：

1. Kafka分区的顺序处理
2. 并行处理每个分区的能力

在试图用这段代码解决这两个问题时，我限制了下游用户配置并行化级别的能力。因此,我更改了该方法,以返回一个GroupedFluxes通量,它为下游用户提供正确的粒度,以确定什么是可并行化的：

public Flux<GroupedFlux<TopicPartition, ConsumerRecord<K, V>>> receiveAuto(Collection<String> topics) {
    return KafkaReceiver.create(createReceiverOptions(topics))
        .receiveAutoAck()
        .flatMap(flux -> flux.groupBy(this::extractTopicPartition));
}

在下游,用户可以使用他们希望的任何调度程序来并行化每个发出的GroupedFlux：

public <V> void work(Flux<GroupedFlux<TopicPartition, V>> flux) {
    flux.doOnNext(groupPublisher -> groupPublisher
            .publishOn(Schedulers.elastic())
            .subscribe(this::doWork))
        .subscribe();
}

这具有所需的行为，按顺序并与其他GroupedFlux并行处理每个TopicPartition-GroupedFlux。