1
# 生产环境可靠性配置清单
2
# Broker 端
3
default.replication.factor=3
4
min.insync.replicas=2
5
unclean.leader.election.enable=false
6

7
# Producer 端
8
acks=all
9
retries=2147483647
10
max.in.flight.requests.per.connection=5
11
enable.idempotence=true
12

13
# Consumer 端
14
enable.auto.commit=false
15
auto.offset.reset=earliest

1
// ACK 策略配置与使用
2
Properties props = new Properties();
3
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
4

5
// acks=0：发完就忘
6
props.put(ProducerConfig.ACKS_CONFIG, "0");
7
// 适用于：日志收集，丢失少量日志可接受
8

9
// acks=1：Leader 确认
10
props.put(ProducerConfig.ACKS_CONFIG, "1");
11
// 适用于：一般业务，Leader 写入即返回
12
// 风险：Leader 确认后宕机，Follower 未同步
13

14
// acks=all：全部 ISR 确认（推荐）
15
props.put(ProducerConfig.ACKS_CONFIG, "all");
16
// 适用于：金融交易、订单等不可丢失场景
17
// 注意：必须配合 min.insync.replicas 使用

1
# min.insync.replicas 配置
2
# Broker 级别（默认值）
3
min.insync.replicas=1
4

5
# Topic 级别（推荐覆盖）
6
kafka-configs --alter --bootstrap-server localhost:9092 \
7
    --entity-type topics --entity-name orders \
8
    --add-config min.insync.replicas=2
9

10
# 验证配置
11
kafka-configs --describe --bootstrap-server localhost:9092 \
12
    --entity-type topics --entity-name orders

1
// 启用幂等生产者
2
Properties props = new Properties();
3
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
4
// 等价于以下配置：
5
// acks=all
6
// retries=Integer.MAX_VALUE
7
// max.in.flight.requests.per.connection<=5
8

9
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
10

11
// 幂等生产者自动为每条消息分配：
12
// 1. PID (Producer ID) - 生产者启动时由 Broker 分配
13
// 2. Sequence Number - 每个 <PID, Topic, Partition> 递增
14
// Broker 端检查：如果收到的 SeqNum <= 已提交的 SeqNum，则判定为重复

1
// 事务生产者配置
2
Properties props = new Properties();
3
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
4
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-tx-1");
5
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
6
props.put(ProducerConfig.ACKS_CONFIG, "all");
7

8
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
9

10
// 初始化事务
11
producer.initTransactions();
12

13
try {
14
    // 开始事务
15
    producer.beginTransaction();
16

17
    // 从 Topic A 消费并处理
18
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
19
    for (ConsumerRecord<String, String> record : records) {
20
        // 处理消息
21
        String result = processOrder(record.value());
22
        // 写入 Topic B
23
        producer.send(new ProducerRecord<>("orders-processed", record.key(), result));
24
    }
25

26
    // 提交消费 Offset（作为事务的一部分）
27
    producer.sendOffsetsToTransaction(
28
        getOffsetsToCommit(records),
29
        consumer.groupMetadata()
30
    );
31

32
    // 提交事务（原子性：写入 + Offset 提交）
33
    producer.commitTransaction();
34
} catch (ProducerFencedException e) {
35
    // 另一个具有相同 transactional.id 的生产者已启动
36
    producer.close();
37
} catch (KafkaException e) {
38
    // 事务失败，回滚
39
    producer.abortTransaction();
40
}

1
// 消费者配置
2
Properties consumerProps = new Properties();
3
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
4
// read_committed 消费者不会返回未提交事务的消息
5
// 它会等待事务提交或超时后才返回消息
6

7
// 事务超时配置
8
// Producer 端
9
props.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "60000"); // 60 秒
10

11
// Broker 端
12
// transaction.max.timeout.ms=900000  // 最大 15 分钟
13
// transactional.id.expiration.ms=604800000  // 7 天过期

1
// 组合提交模式（推荐）
2
try {
3
    while (true) {
4
        ConsumerRecords<String, String> records =
5
            consumer.poll(Duration.ofMillis(100));
6

7
        for (ConsumerRecord<String, String> record : records) {
8
            processRecord(record);
9
        }
10

11
        // 正常情况异步提交（高性能）
12
        consumer.commitAsync();
13
    }
14
} finally {
15
    try {
16
        // 关闭前同步提交（确保最后一次提交成功）
17
        consumer.commitSync();
18
    } finally {
19
        consumer.close();
20
    }
21
}
22

23
// 精确 Offset 提交（处理完每条消息后）
24
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
25
for (ConsumerRecord<String, String> record : records) {
26
    try {
27
        processRecord(record);
28
        offsets.put(
29
            new TopicPartition(record.topic(), record.partition()),
30
            new OffsetAndMetadata(record.offset() + 1)
31
        );
32
    } catch (Exception e) {
33
        // 处理失败，不提交该消息的 Offset
34
        log.error("处理消息失败: offset={}", record.offset(), e);
35
    }
36
}
37
if (!offsets.isEmpty()) {
38
    consumer.commitSync(offsets);
39
}

1
// 消费者重试 + 死信队列模式
2
public class RetryConsumer {
3
    private final KafkaProducer<String, String> dlqProducer;
4
    private final int maxRetries;
5

6
    public void consume() {
7
        while (true) {
8
            ConsumerRecords<String, String> records =
9
                consumer.poll(Duration.ofMillis(100));
10

11
            for (ConsumerRecord<String, String> record : records) {
12
                int retryCount = getRetryCount(record);
13

14
                if (retryCount >= maxRetries) {
15
                    // 超过重试次数，发送到死信队列
16
                    dlqProducer.send(new ProducerRecord<>(
17
                        "orders-dlq",
18
                        record.key(),
19
                        enrichWithRetryInfo(record, retryCount)
20
                    ));
21
                } else {
22
                    try {
23
                        processRecord(record);
24
                    } catch (Exception e) {
25
                        // 重试：发送到重试 Topic
26
                        dlqProducer.send(new ProducerRecord<>(
27
                            "orders-retry",
28
                            record.key(),
29
                            enrichWithRetryInfo(record, retryCount + 1)
30
                        ));
31
                    }
32
                }
33
            }
34
            consumer.commitSync();
35
        }
36
    }
37
}

1
// Kafka Streams exactly-once 配置
2
Properties props = new Properties();
3
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
4
    StreamsConfig.EXACTLY_ONCE_V2);
5
// Kafka 3.0+ 使用 EXACTLY_ONCE_V2（更高效的事务协议）
6

7
// Kafka Streams 内部使用事务 API 保证 exactly-once
8
// 1. 消费输入 Topic
9
// 2. 处理记录
10
// 3. 写入输出 Topic + 更新状态存储 + 提交 Offset
11
// 以上三步在同一个事务中完成
12

13
StreamsBuilder builder = new StreamsBuilder();
14
KStream<String, String> orders = builder.stream("orders");
15

16
orders.groupByKey()
17
    .aggregate(
18
        () -> 0.0,  // 初始值
19
        (key, value, aggregate) -> aggregate + parseAmount(value),
20
        Materialized.with(Serdes.String(), Serdes.Double())
21
    )
22
    .toStream()
23
    .to("order-totals");

1
# 消费者 Lag 监控
2
kafka-consumer-groups --bootstrap-server localhost:9092 \
3
    --describe --group order-service
4

5
# 生产者指标（JMX）
6
# kafka.producer:type=producer-metrics,client-id=*
7
# kafka.producer:type=producer-topic-metrics,client-id=*,topic=*
8

9
# Broker 可靠性指标
10
# kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions
11
# kafka.server:type=ReplicaManager,name=OfflinePartitionsCount
12
# kafka.server:type=TransactionCoordinator,name=*

1
# 模拟 Broker 宕机
2
# 1. 启动 3 Broker 集群
3
# 2. 启动生产者持续发送消息
4
# 3. 杀掉一个 Broker
5
kill -9 <broker-pid>
6

7
# 4. 验证消息不丢失
8
kafka-console-consumer --bootstrap-server localhost:9092 \
9
    --topic orders --from-beginning \
10
    --property print.offset=true
11

12
# 模拟网络分区
13
iptables -A INPUT -s <broker-ip> -j DROP
14
iptables -D INPUT -s <broker-ip> -j DROP
15

16
# 性能测试（可靠性配置下）
17
kafka-producer-perf-test --topic orders \
18
    --num-records 1000000 \
19
    --record-size 1024 \
20
    --throughput -1 \
21
    --producer-props bootstrap.servers=localhost:9092 \
22
        acks=all \
23
        enable.idempotence=true