1
# 查看 Broker 信息
2
kafka-broker-api-versions --bootstrap-server localhost:9092
3

4
# 查看 Controller
5
kafka-metadata --bootstrap-server localhost:9092 --describe
6

7
# Broker 关键配置
8
# server.properties
9
broker.id=1
10
listeners=PLAINTEXT://:9092
11
log.dirs=/var/lib/kafka/data
12
num.network.threads=3
13
num.io.threads=8
14
socket.send.buffer.bytes=102400
15
socket.receive.buffer.bytes=102400

1
// 创建 Topic：3 个分区，3 个副本
2
Properties props = new Properties();
3
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
4

5
try (AdminClient admin = AdminClient.create(props)) {
6
    NewTopic topic = new NewTopic("orders", 3, (short) 3);
7
    admin.createTopics(Collections.singletonList(topic)).all().get();
8
}
9

10
// 查看分区详情
11
DescribeTopicsResult result = admin.describeTopics(
12
    Collections.singletonList("orders")
13
);
14
result.topicNameValues().get("orders").get().partitions()
15
    .forEach(p -> {
16
        System.out.printf("Partition %d: Leader=%d, Replicas=%s, ISR=%s%n",
17
            p.partition(), p.leader().id(),
18
            p.replicas(), p.isr());
19
    });

1
// 分区策略示例
2
// 1. 指定分区
3
ProducerRecord<String, String> record1 =
4
    new ProducerRecord<>("orders", 0, "key1", "value1");
5

6
// 2. Key Hash 分区
7
ProducerRecord<String, String> record2 =
8
    new ProducerRecord<>("orders", "orderId-123", "value2");
9
// Kafka 会计算 hash("orderId-123") % numPartitions
10

11
// 3. 无 Key 轮询
12
ProducerRecord<String, String> record3 =
13
    new ProducerRecord<>("orders", "value3");
14
// StickyPartitioner 会批量发送到同一分区，减少请求数
15

16
// 自定义分区器
17
public class OrderPartitioner implements Partitioner {
18
    @Override
19
    public int partition(String topic, Object key, byte[] keyBytes,
20
                         Object value, byte[] valueBytes, Cluster cluster) {
21
        // 按订单类型分区：普通订单 → 0-2，VIP 订单 → 3-5
22
        String orderType = extractOrderType(value.toString());
23
        int numPartitions = cluster.partitionCountForTopic(topic);
24
        if ("VIP".equals(orderType)) {
25
            return 3 + Math.abs(key.hashCode() % 3);
26
        }
27
        return Math.abs(key.hashCode() % 3);
28
    }
29
}

1
# 增加分区（只能增加，不能减少）
2
kafka-topics --alter --bootstrap-server localhost:9092 \
3
    --topic orders --partitions 6
4

5
# 分区数估算公式
6
# 目标吞吐量 / 单分区吞吐量 = 所需分区数
7
# 例：目标 100MB/s，单分区 10MB/s → 10 个分区
8

9
# 单分区吞吐量基准测试
10
kafka-producer-perf-test --topic test \
11
    --num-records 1000000 \
12
    --record-size 1024 \
13
    --throughput -1 \
14
    --producer-props bootstrap.servers=localhost:9092

1
// 副本分配策略
2
// 手动指定副本分布
3
Map<Integer, List<Integer>> assignment = new HashMap<>();
4
// Partition 0: Leader=0, Followers=1,2
5
assignment.put(0, Arrays.asList(0, 1, 2));
6
// Partition 1: Leader=1, Followers=2,0
7
assignment.put(1, Arrays.asList(1, 2, 0));
8
// Partition 2: Leader=2, Followers=0,1
9
assignment.put(2, Arrays.asList(2, 0, 1));
10

11
admin.createPartitions(
12
    new CreatePartitionsOptions()
13
).call(); // 使用 NewPartitions.increaseTo() 增加分区

1
# 查看 ISR 状态
2
kafka-topics --describe --bootstrap-server localhost:9092 \
3
    --topic orders
4

5
# 输出示例：
6
# Topic: orders  PartitionCount: 3  ReplicationFactor: 3
7
# Topic: orders  Partition: 0  Leader: 1  Replicas: 1,2,3  Isr: 1,2
8
# Topic: orders  Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 2,3,1
9
# Topic: orders  Partition: 2  Leader: 3  Replicas: 3,1,2  Isr: 3,1
10

11
# 关键配置
12
# 生产者 ACK + min.insync.replicas 配合保证可靠性
13
# acks=all + min.insync.replicas=2 → 至少 2 个副本确认才返回成功

1
// Leader 选举过程（简化版）
2
// 1. Controller 检测到 Leader 宕机
3
// 2. 从 ISR 中选择第一个存活的副本作为新 Leader
4
// 3. 更新 ZooKeeper/KRaft 中的元数据
5
// 4. 通知所有 Broker 新的 Leader 信息
6

7
// 使用 AdminClient 查看 Leader Epoch
8
DescribeTopicsResult desc = admin.describeTopics(
9
    Collections.singletonList("orders")
10
);
11
// LeaderAndIsr 包含：leader, leaderEpoch, isr, zkVersion

1
// 消费者组配置
2
Properties props = new Properties();
3
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
4
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service");
5
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
6
    StringDeserializer.class.getName());
7
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
8
    StringDeserializer.class.getName());
9
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
10
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
11
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500");
12
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "300000");
13

14
KafkaConsumer<String, String> consumer =
15
    new KafkaConsumer<>(props);
16
consumer.subscribe(Collections.singletonList("orders"));

1
// RangeAssignor 示例
2
// Topic: orders, 7 个分区, 3 个消费者
3
// Consumer 0: Partition 0, 1, 2  (3个)
4
// Consumer 1: Partition 3, 4     (2个)
5
// Consumer 2: Partition 5, 6     (2个)
6

7
// RoundRobinAssignor 示例
8
// Topic: orders, 7 个分区, 3 个消费者
9
// Consumer 0: Partition 0, 3, 6  (3个)
10
// Consumer 1: Partition 1, 4     (2个)
11
// Consumer 2: Partition 2, 5     (2个)
12

13
// 设置分配策略
14
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
15
    StickyAssignor.class.getName());
16

17
// 使用 CooperativeStickyAssignor（推荐）
18
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
19
    CooperativeStickyAssignor.class.getName());

1
# 查看消费者组状态
2
kafka-consumer-groups --bootstrap-server localhost:9092 \
3
    --describe --group order-service
4

5
# 输出示例：
6
# GROUP           TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
7
# order-service   orders  0          1000            1050            50
8
# order-service   orders  1          980             1020            40
9
# order-service   orders  2          950             1000            50
10

11
# 重置消费者组 Offset
12
kafka-consumer-groups --bootstrap-server localhost:9092 \
13
    --group order-service --topic orders \
14
    --reset-offsets --to-earliest --execute

1
// 手动提交 Offset
2
while (true) {
3
    ConsumerRecords<String, String> records =
4
        consumer.poll(Duration.ofMillis(100));
5

6
    for (ConsumerRecord<String, String> record : records) {
7
        // 处理消息
8
        processOrder(record.value());
9
    }
10

11
    // 同步提交（阻塞等待确认）
12
    consumer.commitSync();
13

14
    // 或异步提交（不等待确认）
15
    consumer.commitAsync(new OffsetCommitCallback() {
16
        @Override
17
        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets,
18
                               Exception exception) {
19
            if (exception != null) {
20
                log.error("提交 Offset 失败", exception);
21
            }
22
        }
23
    });
24
}
25

26
// 精确提交 Offset（处理完每条消息后）
27
Map<TopicPartition, OffsetAndMetadata> commitMap = new HashMap<>();
28
for (ConsumerRecord<String, String> record : records) {
29
    processOrder(record.value());
30
    commitMap.put(
31
        new TopicPartition(record.topic(), record.partition()),
32
        new OffsetAndMetadata(record.offset() + 1)
33
    );
34
}
35
consumer.commitSync(commitMap);

1
# Controller 相关操作
2
# 查看当前 Controller
3
kafka-metadata --bootstrap-server localhost:9092 --describe
4

5
# 手动触发 Preferred Leader 选举
6
kafka-leader-election --bootstrap-server localhost:9092 \
7
    --election-type preferred \
8
    --topic orders --partition 0
9

10
# 分区重分配
11
cat > reassignment.json << 'EOF'
12
{
13
    "partitions": [
14
        {"topic": "orders", "partition": 0, "replicas": [3, 1, 2]},
15
        {"topic": "orders", "partition": 1, "replicas": [1, 2, 3]}
16
    ]
17
}
18
EOF
19

20
kafka-reassign-partitions --bootstrap-server localhost:9092 \
21
    --reassignment-json-file reassignment.json \
22
    --execute

1
# KRaft 模式初始化
2
kafka-storage format -t $(uuidgen) -c server.properties
3

4
# KRaft 模式启动
5
kafka-server-start.sh server.properties
6

7
# KRaft 元数据查询
8
kafka-metadata-quorum --bootstrap-server localhost:9092 describe --status

1
# 生产环境推荐配置
2
# 副本数 = 3（平衡可靠性与成本）
3
# min.insync.replicas = 2（允许 1 个副本故障）
4
# acks = all（确保消息写入所有 ISR 副本）
5

6
# Broker 配置
7
default.replication.factor=3
8
min.insync.replicas=2
9
unclean.leader.election.enable=false
10

11
# Producer 配置
12
acks=all
13
retries=2147483647
14
max.in.flight.requests.per.connection=5
15
enable.idempotence=true
16

17
# Consumer 配置
18
enable.auto.commit=false
19
auto.offset.reset=earliest
20
max.poll.records=500
21
max.poll.interval.ms=300000
22
session.timeout.ms=30000

1
// 使用 Kafka Metrics API 监控
2
Map<MetricName, ? extends Metric> metrics = kafkaConsumer.metrics();
3
metrics.forEach((name, metric) -> {
4
    if (name.name().equals("records-lag-max")) {
5
        System.out.printf("Consumer Lag: %s%n", metric.metricValue());
6
    }
7
});
8

9
// JMX 监控关键 MBean
10
// kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions
11
// kafka.controller:type=KafkaController,name=ActiveControllerCount
12
// kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*