如何通过自动确认按主题和分区同时处理 Reactor Kafka 流？

Question

我正在尝试使用带有自动确认功能的 Reactor Kafka 来实现 Kafka 主题分区的并发处理。这里的文档使这看起来是可能的：

http://projectreactor.io/docs/kafka/milestone/reference/#concurrent-ordered

这与我正在尝试的唯一区别是我使用的是自动确认。

我有如下代码（相关方法为receiveAuto）：

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());
    }
}

当我使用它通过并行调度程序 (Schedulers.newParallel("debug", 10)) 从 Kafka 创建消费者记录通量时，我看到它们最终都在同一个线程上得到处理。

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

Answer 1

经过相当多的反复试验以及对我想要完成的任务的重新思考后，我意识到我试图用一段代码解决两个问题。

我需要的两件事是：

1. 按顺序处理 Kafka 分区
2. 能够并行处理每个分区

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

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();
}

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

Answer 2

我猜它至少在你的消费者中是按顺序执行的。要进行并行消耗，您应该将通量转换为ParallelFlux

public ParallelFlux<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.parallel().runOn(Schedulers.parallel()));
    }

在你的消费者函数之后，如果你想以并行方式消费，你应该使用如下方法：

void subscribe(Consumer<? super T> onNext, Consumer<? super Throwable>
            onError, Runnable onComplete, Consumer<? super Subscription> onSubscribe)

或任何其他带有Consumer<T super T> onNext 参数的重载方法。 如果您只使用以下方法，您将按顺序消耗通量

void subscribe(Subscriber<? super T> s)