如何在Spring Reactor Kafka中创建多个KafkaReceiver实例

你当前的实现确实可以创建多个消费者实例并行消费分区，属于可用实现，但并非最高效，存在几个可优化的问题：

• 硬编码分区数量：直接绑定topicPartitionsCount变量，后续topic扩容分区时需要修改代码重启应用，无法自动适配
• 消息可靠性不足：使用receiveAutoAck()会在消息被消费者接收后立即提交偏移量，和下游发送逻辑完全解绑，若发送到目标topic失败会直接丢失消息
• 背压脱节：sendToKafka方法中单独订阅发送流，消费和发送之间没有背压传递，若生产速度跟不上消费速度，会导致大量消息积压在内存中引发OOM
• 资源利用率不稳定：若部署多个应用实例，同一个消费者组下总消费者数量超过topic分区数时，多余的消费者实例会处于闲置状态，浪费系统资源

方案1：动态适配分区数的订阅模式 #

无需硬编码分区数量，应用启动时通过Kafka AdminClient自动获取目标topic的实际分区数，再创建对应数量的消费流，同时整合消费和发送逻辑，手动控制偏移量提交保证消息可靠性，示例代码如下：

// 首先注入AdminClient获取分区数
@Autowired
private AdminClient adminClient;

public void run(String... args) {
    // 动态获取topic分区数
    int partitionCount = adminClient.describeTopics(Collections.singletonList(sourceTopic))
            .allTopicNames().block().get(sourceTopic).partitions().size();
    // 最多创建和分区数一致的流
    for(int i = 0; i < partitionCount ; i++) {
        readWrite(destinationTopic).subscribe();
    }
}

public Flux<String> readWrite(String destTopic) {
    return kafkaConsumerTemplate
            // 用receive方法手动控制偏移量提交，不要用自动ack
            .receive()
            .doOnNext(consumerRecord -> log.info("received key={}, value={} from topic={}, offset={}",
                    consumerRecord.key(),
                    consumerRecord.value(),
                    consumerRecord.topic(),
                    consumerRecord.offset())
            )
            // 用flatMap整合发送逻辑，打通背压，发送成功再提交偏移量
            .flatMap(consumerRecord -> kafkaProducerTemplate.send(destTopic, consumerRecord.key(), transformRecord(consumerRecord))
                    .doOnSuccess(senderResult -> {
                        log.debug("Sent record offset : {}", senderResult.recordMetadata().offset());
                        // 发送成功后手动提交偏移量
                        consumerRecord.receiverOffset().acknowledge();
                    })
                    .doOnError(exception -> {
                        log.error("Error while sending message to destination topic : {}", exception.getMessage());
                    })
            )
            .map(senderResult -> senderResult.correlationMetadata().toString())
            .onErrorContinue((exception, record)->{
                log.error("Error while processing : {}", exception.getMessage());
            });
}

方案2：手动分配分区模式（单实例部署推荐） #

如果你的应用是单实例部署，不需要考虑多实例rebalance，可以直接手动给每个Receiver分配固定分区，避免rebalance带来的开销，性能比订阅模式高10%~15%：

// 为每个分区创建独立的ReceiverOptions，指定分配的分区
public List<ReceiverOptions<String, String>> buildPartitionAssignOptions(String topic, int partitionCount, KafkaProperties kafkaProperties) {
    List<ReceiverOptions<String, String>> optionsList = new ArrayList<>();
    for (int i = 0; i < partitionCount; i++) {
        TopicPartition tp = new TopicPartition(topic, i);
        ReceiverOptions<String, String> options = ReceiverOptions.create(kafkaProperties.buildConsumerProperties())
                .assignment(Collections.singleton(tp));
        optionsList.add(options);
    }
    return optionsList;
}

// 启动时为每个分区创建独立的消费流
public void run(String... args) {
    List<ReceiverOptions<String, String>> optionsList = buildPartitionAssignOptions(sourceTopic, partitionCount, kafkaProperties);
    optionsList.forEach(options -> {
        new ReactiveKafkaConsumerTemplate<String, String>(options)
                .receive()
                // 后续消费发送逻辑和方案1一致
                .flatMap(this::processAndSend)
                .subscribe();
    });
}

• 可以通过ReceiverOptions.maxConcurrency()配置单消费者实例的内部处理并发，不需要完全靠增加消费者实例数量提升处理能力
• 若需要部署多实例，不要在每个实例内创建和分区数一致的消费者，建议统一配置每个实例启动N个消费者，总消费者数量等于分区数即可，避免资源浪费

内容的提问来源于stack exchange，提问作者perplexedDev