Reactor Kafka 非阻塞背压机制实战指南

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.util.Collections;
import java.util.HashMap;
import java.util.Map;

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-consumer-group";

    public static void main(String[] args) {
        // 1. 配置Kafka消费者属性
        Map<String, Object> consumerProps = new HashMap<>();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // 自动提交offset设置为false，由Reactor Kafka手动提交或在处理完成后提交
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 设置每次poll的最大记录数，这也是一种粗粒度的背压控制
        consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500); // 每次最多拉取500条消息

        // 2. 创建ReceiverOptions对象，配置Kafka接收器
        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.<String, String>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);

        // 4. 使用flatMap操作符处理消息并应用背压
        kafkaReceiver.receive() // 获取消息流
                .flatMap(record -> {
                    // 模拟消息处理逻辑，例如调用外部服务、数据库操作等
                    System.out.println("接收到消息 - Topic: " + record.topic() +
                            ", Partition: " + record.partition() +
                            ", Offset: " + record.offset() +
                            ", Key: " + record.key() +
                            ", Value: " + record.value());

                    // 模拟耗时操作，例如100毫秒
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }

                    // 返回一个Mono<Void>表示当前消息处理完成。
                    // 当这个Mono完成时，Reactor会知道可以请求下一个消息。
                    // 如果这里返回一个包含实际结果的Mono，后续操作可以继续处理结果。
                    // 对于仅表示处理完成的场景，Mono.empty()是合适的。
                    return Mono.empty();
                })
                .doOnError(e -> System.err.println("处理消息时发生错误: " + e.getMessage())) // 错误处理
                .subscribe(); // 订阅并开始消费消息

        // 通常，在实际应用中，你需要一个机制来保持主线程运行，
        // 例如使用CountDownLatch或Spring Boot应用上下文。
        // 这里只是一个简单的示例，程序会在消息处理完成后退出（如果消息流有限）。
        // 对于无限流，它会持续运行。
        System.out.println("Kafka消费者已启动，正在等待消息...");
        // 为了演示，这里可以添加一个阻塞，防止主线程立即退出
        try {
            Thread.currentThread().join(); 
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}