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

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.logging.Logger;

public class ReactiveKafkaBackpressureExample {

    private static final Logger logger = Logger.getLogger(ReactiveKafkaBackpressureExample.class.getName());
    private static final String BOOTSTRAP_SERVERS = "localhost:9092"; // Kafka服务器地址
    private static final String TOPIC = "my-reactive-topic";         // 订阅的Kafka主题
    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(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，由我们手动控制提交，以便更好地实现反压和容错
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // 每次poll的最大记录数，可以与Reactor的反压机制协同工作
        consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500");

        // 2. 创建ReceiverOptions，配置订阅主题和监听器
        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.<String, String>create(consumerProps)
                .subscription(Collections.singleton(TOPIC))
                .addAssignListener(partitions -> logger.info("onPartitionsAssigned: " + partitions))
                .addRevokeListener(partitions -> logger.info("onPartitionsRevoked: " + partitions));

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

        // 4. 使用flatMap操作符实现反压和异步处理
        // flatMap的第二个参数concurrency可以控制同时处理的Mono/Flux数量，
        // 达到上限时，flatMap会暂停从上游（KafkaReceiver）拉取新消息，从而实现反压。
        kafkaReceiver.receive()
                .flatMap(record -> {
                    // 模拟消息处理，例如：数据库写入、外部API调用等耗时操作
                    // 这里使用Mono.delay模拟一个耗时操作，每个消息处理耗时100毫秒
                    logger.info("开始处理消息: " + record.key() + " -> " + record.value() + " (Partition: " + record.partition() + ", Offset: " + record.offset() + ")");
                    return Mono.delay(Duration.ofMillis(100))
                            .doOnSuccess(v -> {
                                logger.info("消息处理完成: " + record.key());
                                // 消息处理成功后，手动提交offset
                                // 在实际应用中，通常会批量提交或在事务中提交
                                record.receiverOffset().commit()
                                      .doOnError(e -> logger.severe("提交offset失败: " + e.getMessage()))
                                      .subscribe(); // 提交操作也应是非阻塞的
                            })
                            .doOnError(e -> logger.severe("处理消息失败: " + record.key() + " - " + e.getMessage()))
                            .thenReturn(record); // 返回处理后的record，表示该消息已处理
                }, 5) // 设置并发度为5，意味着最多同时处理5个消息
                .subscribe(
                        record -> logger.info("消费者成功订阅并处理消息流中的一个元素。"),
                        error -> logger.severe("消费者流发生错误: " + error.getMessage()),
                        () -> logger.info("消费者流完成。") // 通常Kafka消费者是无限流，不会完成
                );

        // 保持主线程运行，以便KafkaReceiver可以持续接收消息
        Thread.currentThread().join();
    }
}