深入理解Reactive Kafka的非阻塞背压机制及Java实现

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import reactor.core.Disposable;
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; // 生产环境建议使用SLF4J等日志框架

public class ReactiveKafkaBackpressureConsumer {

    private static final Logger LOGGER = Logger.getLogger(ReactiveKafkaBackpressureConsumer.class.getName());
    private static final String KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"; // 替换为您的Kafka地址
    private static final String KAFKA_TOPIC = "my-reactive-topic";
    private static final String KAFKA_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, KAFKA_BOOTSTRAP_SERVERS);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, KAFKA_GROUP_ID);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从最早的偏移量开始消费
        // 禁用自动提交偏移量，由Reactive Kafka手动管理
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        // 2. 配置ReceiverOptions
        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.<String, String>create(consumerProps)
                .subscription(Collections.singleton(KAFKA_TOPIC)) // 订阅指定主题
                .addAssignListener(partitions -> LOGGER.info("分区分配: " + partitions))
                .addRevokeListener(partitions -> LOGGER.info("分区撤销: " + partitions));

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

        // 4. 消费消息并应用背压
        LOGGER.info("开始消费Kafka消息...");
        Disposable disposable = kafkaReceiver.receive()
                // flatMap操作符是实现背压的关键。
                // 它允许并发处理消息，但通过其第二个参数（concurrency）限制并发数。
                // 当并发处理的消息达到上限时，flatMap会向上游（KafkaReceiver）发出背压信号，
                // 暂停拉取新消息，直到有处理完成的槽位。
                .flatMap(record -> processMessage(record)
                                .doOnSuccess(v -> {
                                    // 消息成功处理后，手动确认偏移量
                                    record.receiverOffset().acknowledge();
                                    LOGGER.info("消息处理并确认: Key=" + record.key() + ", Value=" + record.value() + ", Offset=" + record.offset());
                                })
                                .doOnError(e -> {
                                    LOGGER.severe("处理消息失败: Key=" + record.key() + ", Error=" + e.getMessage());
                                    // 错误处理策略：可以重试、记录日志、将消息发送到死信队列等
                                    // 注意：这里没有调用acknowledge()，意味着该消息可能在下次消费时再次被处理
                                }),
                        10) // 并发处理消息的最大数量，这是背压的关键控制点
                .doOnError(e -> LOGGER.severe("Kafka消息流发生错误: " + e.getMessage()))
                .doOnTerminate(() -> LOGGER.info("Kafka消费者流终止。"))
                .subscribe(); // 订阅Flux以启动消费流程

        // 在实际应用中，您会管理disposable的生命周期，例如在应用程序关闭时调用disposable.dispose()
        // 这里为了演示，我们让主线程等待一段时间或者通过其他方式保持运行
        try {
            Thread.sleep(Long.MAX_VALUE); // 保持主线程运行，以便消费者持续工作
        } catch (InterruptedException e) {
            LOGGER.info("消费者线程被中断。");
            disposable.dispose(); // 停止消费
        }
    }

    /**
     * 模拟异步消息处理逻辑。
     * 在实际应用中，这可能涉及数据库操作、外部API调用等耗时任务。
     *
     * @param record 接收到的Kafka消息记录
     * @return 返回一个Mono<Void>表示消息处理完成。
     */
    private static Mono<Void> processMessage(ReceiverRecord<String, String> record) {
        // 模拟一个耗时100毫秒的异步操作
        return Mono.delay(Duration.ofMillis(100))
                .doOnSuccess(v -> LOGGER.fine("内部处理完成: Key=" + record.key()))
                .then(); // 转换为Mono<Void>
    }
}