Kafka 分区机制

Topic: orders
    │
    ├── Partition 0: [msg0, msg3, msg6, ...]
    ├── Partition 1: [msg1, msg4, msg7, ...]
    └── Partition 2: [msg2, msg5, msg8, ...]

// 生产者发送消息
ProducerRecord<String, String> record = 
    new ProducerRecord<>("topic", key, value);
 
// 分区计算逻辑
public int partition(String topic, Object key, byte[] keyBytes, 
                     Object value, byte[] valueBytes, Cluster cluster) {
    // 1. 指定分区 → 使用指定分区
    // 2. 有 key → hash(key) % partitionCount
    // 3. 无 key → sticky partitioner（黏性分区器）
}

public class OrderPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int partitionCount = cluster.partitionCountForTopic(topic);
        
        // 按订单类型分区
        if (key instanceof String) {
            String orderKey = (String) key;
            if (orderKey.startsWith("VIP_")) {
                return 0;  // VIP 订单到分区 0
            }
        }
        
        return Math.abs(key.hashCode()) % partitionCount;
    }
    
    @Override
    public void configure(Map<String, ?> configs) {}
    
    @Override
    public void close() {}
}

Topic (3 个分区)
    │
    │    消费者组 A
    │    ├── Consumer-1: 消费 P0, P1
    │    └── Consumer-2: 消费 P2
    │
    │    消费者组 B
    ├── Consumer-3: 消费 P0
    ├── Consumer-4: 消费 P1
    └── Consumer-5: 消费 P2

分区: [P0, P1, P2, P3, P4, P5]
消费者: [C0, C1, C2]

分配结果:
C0: P0, P1  (2个)
C1: P2, P3  (2个)
C2: P4, P5  (2个)

分区: [P0, P1, P2, P3, P4, P5]
消费者: [C0, C1, C2]

分配结果:
C0: P0, P3
C1: P1, P4
C2: P2, P5

Partition 0:
    ┌─────────────────────────────────┐
    │ Leader (Broker 1)               │ ← 读写请求
    │ [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]  │
    └─────────────────────────────────┘
              ↑ 同步
    ┌─────────┴─────────┐
    ↓                   ↓
┌───────────┐     ┌───────────┐
│ Follower  │     │ Follower  │
│ (Broker 2)│     │ (Broker 3)│
└───────────┘     └───────────┘

// ISR 配置
min.insync.replicas=2     // 最小同步副本数
unclean.leader.election.enable=false  // 禁止非 ISR 副本当选 Leader

// 生产者 ACK 配置
props.put("acks", "all");  // 0, 1, all(-1)

Partition 0: [msg1, msg2, msg3, msg4]
              消费顺序: msg1 → msg2 → msg3 → msg4

Partition 0: [msg1, msg3]
Partition 1: [msg2, msg4]

可能消费顺序: msg1 → msg2 → msg3 → msg4
          或: msg2 → msg1 → msg4 → msg3

// 所有消息发到同一分区
new ProducerRecord<>("topic", 0, key, value);  // 指定分区 0

// 相同 key 的消息到同一分区
new ProducerRecord<>("topic", "order-123", value);  // key 相同

1. 消费者发送 JoinGroup 请求
2. Coordinator 选择 Leader 消费者
3. Leader 制定分区分配方案
4. 所有消费者收到分配结果
5. 开始消费

// 心跳配置
session.timeout.ms=10000      // 会话超时时间
heartbeat.interval.ms=3000    // 心跳间隔
max.poll.interval.ms=300000   // 两次 poll 最大间隔
 
// 处理时间过长会导致 Rebalance
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        process(record);  // 单条处理时间不宜过长
    }
}

分区数 = max(
    目标吞吐量 / 单个分区最大吞吐量,
    目标吞吐量 / 单个消费者最大吞吐量
)

分区数 = max(100/20, 100/10) = max(5, 10) = 10

生产者:
- acks=all
- retries=3
- enable.idempotence=true

Broker:
- replication.factor>=3
- min.insync.replicas>=2
- unclean.leader.election.enable=false

消费者:
- enable.auto.commit=false
- 手动提交 offset

props.put("enable.idempotence", true);  // 开启幂等

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 3);
props.put("enable.idempotence", true);
props.put("compression.type", "lz4");
props.put("batch.size", 16384);
props.put("linger.ms", 5);

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "consumer-group");
props.put("enable.auto.commit", false);
props.put("auto.offset.reset", "earliest");
props.put("max.poll.records", 500);
props.put("session.timeout.ms", 10000);