深入理解与实现 Reactor Kafka 的非阻塞背压机制

import org.apache.kafka.common.serialization.StringDeserializer;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverRecord;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

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

public class ReactiveKafkaBackpressureExample {

    private static final String BOOTSTRAP_SERVERS = "localhost:9092"; // Kafka集群地址
    private static final String TOPIC = "my-topic";                  // 订阅主题
    private static final String GROUP_ID = "my-reactive-group";      // 消费者组ID

    public static void main(String[] args) throws InterruptedException {
        // 1. 配置Kafka消费者属性
        Map<String, Object> consumerProps = new HashMap<>();
        consumerProps.put("bootstrap.servers", BOOTSTRAP_SERVERS);
        consumerProps.put("group.id", GROUP_ID);
        consumerProps.put("key.deserializer", StringDeserializer.class.getName());
        consumerProps.put("value.deserializer", StringDeserializer.class.getName());
        consumerProps.put("auto.offset.reset", "earliest"); // 从最早的偏移量开始消费
        // 禁用自动提交，以便手动控制偏移量提交
        consumerProps.put("enable.auto.commit", "false"); 

        // 2. 创建ReceiverOptions，配置订阅主题和监听器
        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(consumerProps)
                .subscription(Collections.singleton(TOPIC)) // 订阅指定主题
                .addAssignListener(partitions -> System.out.println("分区分配: " + partitions))
                .addRevokeListener(partitions -> System.out.println("分区撤销: " + partitions));

        // 3. 创建KafkaReceiver实例
        KafkaReceiver<String, String> kafkaReceiver = KafkaReceiver.create(receiverOptions);

        // 使用CountDownLatch来控制程序的生命周期，以便观察效果
        CountDownLatch latch = new CountDownLatch(1);

        System.out.println("开始消费Kafka消息...");

        // 4. 使用flatMap操作符实现非阻塞背压
        kafkaReceiver.receive() // 获取Flux<ReceiverRecord>流
                .publishOn(Schedulers.parallel()) // 将消息处理调度到并行线程池，避免阻塞Kafka I/O线程
                .flatMap(record -> {
                    // 模拟消息处理，这里可以是异步操作，例如数据库写入、调用外部服务等
                    System.out.printf("接收到消息: Key=%s, Value=%s, Offset=%d, Partition=%d%n",
                                      record.key(), record.value(), record.offset(), record.partition());

                    // 模拟一个耗时操作，例如处理业务逻辑或外部服务调用
                    return Mono.delay(Duration.ofMillis(500)) // 模拟耗时500毫秒
                               .doOnSuccess(v -> {
                                   // 消息处理完成后，手动提交偏移量
                                   record.receiverOffset().acknowledge(); // 确认消息已被处理
                                   System.out.printf("消息处理完成并提交偏移量: Offset=%d%n", record.offset());
                               })
                               .then(Mono.empty()); // 表示当前消息的处理已完成，不向下游发出新元素
                }, 3) // flatMap的concurrency参数，控制同时处理的消息数量，实现背压
                .doOnError(e -> System.err.println("处理消息时发生错误: " + e.getMessage()))
                .doOnCancel(() -> System.out.println("Kafka接收器被取消"))
                .doFinally(signalType -> {
                    System.out.println("Kafka接收器完成或终止: " + signalType);
                    latch.countDown(); // 释放锁，允许主线程退出
                })
                .subscribe(
                    null, // onNext: 因为flatMap返回Mono.empty()，所以这里不会有元素发出
                    error -> System.err.println("Kafka接收器订阅链发生错误: " + error),
                    () -> System.out.println("Kafka接收器完成（通常不会发生，除非流被显式完成）")
                );

        // 阻止主线程立即退出，以便KafkaReceiver持续运行
        latch.await();
        System.out.println("程序退出。");
    }
}