Reactive Kafka非阻塞背压机制在Java中的实现与应用

<dependency>
    <groupId>io.projectreactor.kafka</groupId>
    <artifactId>reactor-kafka</artifactId>
    <version>1.3.18</version> <!-- 使用最新稳定版本 -->
</dependency>
<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>3.5.15</version> <!-- 使用最新稳定版本 -->
</dependency>

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import reactor.core.publisher.Mono;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.receiver.ReceiverRecord;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class ReactiveKafkaBackpressureExample {

    public static void main(String[] args) throws InterruptedException {
        // 1. Kafka消费者配置
        Map<String, Object> consumerProps = new HashMap<>();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // 替换为你的Kafka地址
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "my-reactive-group");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 禁用自动提交，由Reactive Kafka手动管理偏移量，以实现精确的背压和消息确认
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 调整max.poll.records，避免一次拉取过多消息，与flatMap的并发度协同工作
        consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100); 

        // 2. 配置Kafka接收器选项
        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.<String, String>create(consumerProps)
                .subscription(Collections.singleton("my-topic")) // 订阅主题
                .addAssignListener(partitions -> System.out.println("分区分配: " + partitions))
                .addRevokeListener(partitions -> System.out.println("分区撤销: " + partitions));

        // 3. 创建Kafka接收器
        KafkaReceiver<String, String> kafkaReceiver = KafkaReceiver.create(receiverOptions);

        CountDownLatch latch = new CountDownLatch(1); // 用于演示程序运行直到终止

        System.out.println("开始消费Kafka消息，应用非阻塞背压...");

        // 4. 使用flatMap操作符实现背压和消息处理
        kafkaReceiver.receive() // 获取Flux<ReceiverRecord<K, V>>
                // flatMapWithConcurrency: 指定并发处理消息的数量。
                // 例如，concurrency = 2 意味着最多同时处理2条消息。
                // 当2条消息都在处理中时，KafkaReceiver将暂停从Kafka拉取新消息，直到有处理完成。
                .flatMap(record -> processAndCommit(record), 2) // 设置并发度为2，模拟背压
                .doOnError(e -> System.err.println("消息处理错误: " + e.getMessage())) // 捕获并打印处理过程中的错误
                .doOnTerminate(() -> {
                    System.out.println("Kafka消费者终止。");
                    latch.countDown();
                })
                .subscribe(
                        null, // 不关心每个内部Mono的完成信号，我们只关心整个流的错误和完成
                        error -> System.err.println("流整体错误: " + error), // 捕获整个流的错误
                        () -> System.out.println("流完成。") // 流正常完成（通常不会发生，除非Kafka关闭）
                );

        // 保持主线程运行，直到有信号终止，或达到超时
        latch.await(10, TimeUnit.MINUTES);
    }

    /**
     * 模拟消息处理并提交偏移量
     * @param record 接收到的Kafka消息记录
     * @return 返回一个Mono<Void>，表示处理完成
     */
    private static Mono<Void> processAndCommit(ReceiverRecord<String, String> record) {
        return Mono.defer(() -> {
            System.out.println(String.format("收到消息 -> Topic: %s, Partition: %d, Offset: %d, Key: %s, Value: %s",
                    record.topic(), record.partition(), record.offset(), record.key(), record.value()));

            // 模拟一个耗时的业务处理。Mono.delay是非阻塞的。
            return Mono.delay(Duration.ofMillis(500)) // 模拟500ms的异步处理
                    .doOnNext(x -> System.out.println("处理完成 -> Offset: " + record.offset()))
                    .then(Mono.defer(() -> {
                        // 提交偏移量。只有在消息处理成功后才提交。
                        record.receiverOffset().acknowledge();
                        System.out.println("提交偏移量 -> Offset: " + record.offset());
                        return Mono.empty(); // 返回一个完成信号，表示此消息的处理流程已结束
                    }));
        });
    }
}