Kafka max.in.flight.requests.per.connection与Reactor Kafka maxInflight差异及用法咨询

核心差异 #

1. 作用层级与控制维度不同 #

• 原生max.in.flight.requests.per.connection是Kafka Java客户端底层的配置，控制单个Broker连接上同时未收到ACK的请求数量，每个请求对应一次向Broker发起的生产操作（可能包含一个或多个消息批次）。
• Reactor Kafka的maxInflight是反应式封装层的限流配置，控制发送订阅流中同时处于"已发送但未完成ACK"状态的消息批次总数，和底层连接数无关，是Reactor响应式流层面的背压控制手段。

2. 独立性 #

两者完全独立生效：修改Reactor的maxInflight不会影响原生客户端的max.in.flight.requests.per.connection配置，反之亦然。maxInflight相当于在原生客户端之上额外加了一层流控逻辑。

3. 适用场景侧重不同 #

• 原生配置主要用于控制底层网络连接的并发请求数，避免Broker端压力过载或客户端请求超时；若设置为1，可严格保证单分区的消息顺序（前一个请求ACK后才会发送下一个）。
• Reactor的maxInflight主要用于控制响应式流中的未处理批次数量，避免客户端内存中堆积过多未ACK的消息，适配反应式编程的背压机制，平衡吞吐量与内存占用。

具体使用示例 #

1. Kafka原生Producer配置示例 #

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;

public class NativeProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 设置底层单个连接的最大未ACK请求数
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        // 发送消息逻辑（省略）
    }
}

2. Reactor Kafka Sender配置与使用示例 #

import reactor.kafka.sender.KafkaSender;
import reactor.kafka.sender.SenderOptions;
import reactor.kafka.sender.SenderRecord;
import reactor.core.publisher.Flux;

import java.util.Properties;

public class ReactorKafkaExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 保留原生底层配置
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

        // 创建SenderOptions并设置Reactor层面的maxInflight
        SenderOptions<String, String> senderOptions = SenderOptions.<String, String>create(props)
                .maxInflight(3); // 限制同时处于飞行状态的消息批次为3个

        KafkaSender<String, String> sender = KafkaSender.create(senderOptions);

        // 构造待发送的消息流
        Flux<SenderRecord<String, String, String>> records = Flux.just(
                SenderRecord.create("test-topic", null, null, "key1", "value1", "corr-1"),
                SenderRecord.create("test-topic", null, null, "key2", "value2", "corr-2"),
                SenderRecord.create("test-topic", null, null, "key3", "value3", "corr-3"),
                SenderRecord.create("test-topic", null, null, "key4", "value4", "corr-4")
        );

        // 发送并处理ACK结果
        sender.send(records)
                .doOnNext(result -> System.out.println("消息ACK完成，关联ID: " + result.correlationMetadata()))
                .subscribe();
    }
}

示例说明：此例中，原生配置允许单个连接同时有5个未ACK请求，但Reactor的maxInflight=3会限制发送流中最多同时存在3个未ACK的批次。只有当某个批次收到ACK后，Reactor才会继续发送第4个批次，避免流中堆积过多未处理消息。

内容的提问来源于stack exchange，提问作者RamPrakash