Kafka分区策略与消费者组重平衡机制生产实战

Kafka作为高吞吐消息中间件，其并行处理能力依赖于分区（Partition）机制。生产者将消息写入分区，消费者组内每个消费者负责不同分区的消息消费。分区的数量和分配策略直接决定Kafka集群的吞吐量和扩展性。后端开发中，理解Kafka分区策略与消费者组重平衡机制，对设计高并发消息处理系统和排查消费延迟问题至关重要。

Kafka分区分配策略与分区数规划

Kafka Topic由多个分区组成，每个分区是一个有序的、不可变的消息序列。消息通过分区器（Partitioner）决定写入哪个分区。默认分区器根据消息key的hash值分配分区，无key时采用轮询（Round-Robin）策略。

// Java生产者自定义分区器
public class CustomPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();

        if (keyBytes == null) {
            // 无key：按value hash分配，保证相同内容路由到同一分区
            return Math.abs(Utils.murmur2(valueBytes)) % numPartitions;
        }

        String keyStr = new String(keyBytes, StandardCharsets.UTF_8);

        // 按用户ID分区：保证同一用户消息有序
        if (keyStr.startsWith("user:")) {
            int userIdHash = Math.abs(keyStr.hashCode());
            return userIdHash % numPartitions;
        }

        // 按地域分区：将同地域消息路由到指定分区
        if (keyStr.startsWith("region:")) {
            String region = keyStr.substring(7);
            return getRegionPartition(region, numPartitions);
        }

        return Math.abs(Utils.murmur2(keyBytes)) % numPartitions;
    }
}

// 生产者配置
Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("partitioner.class", "com.example.CustomPartitioner");
props.put("acks", "all");
props.put("retries", 3);
props.put("batch.size", 16384);
props.put("linger.ms", 10);
props.put("compression.type", "lz4");

分区数规划需考虑三个因素：目标吞吐量、消费者并行度和broker磁盘性能。单个分区的写入吞吐受限于磁盘IO，通常约10MB/s。若目标吞吐量100MB/s，至少需要10个分区。消费者组中活跃消费者数量不能超过分区数，否则多余消费者空闲。

# 分区数计算公式
# 生产者吞吐: partition_producer_throughput = min(producer_max, disk_write_bw)
# 消费者吞吐: partition_consumer_throughput = min(consumer_max, disk_read_bw)
# 所需分区数 = max(
#   ceil(target_producer_throughput / partition_producer_throughput),
#   ceil(target_consumer_throughput / partition_consumer_throughput)
# )

# 创建Topic时指定分区数
kafka-topics.sh --create   --topic order-events   --partitions 24   --replication-factor 3   --bootstrap-server kafka:9092

# 分区数不宜过多：每个分区消耗broker内存和文件句柄
# 经验值：单broker分区总数建议不超过4000

消费者组重平衡机制与Rebalance协议

消费者组重平衡（Rebalance）是指消费者组成员变化时，重新分配分区与消费者的映射关系。触发重平衡的场景包括：消费者加入或离开组、消费者心跳超时、订阅Topic分区数变化。重平衡期间消费者无法消费消息，造成消费暂停（Stop-the-World）。

// 消费者配置
Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("group.id", "order-consumer-group");
props.put("enable.auto.commit", "false");
props.put("auto.offset.reset", "earliest");
props.put("max.poll.records", 500);
props.put("max.poll.interval.ms", 300000);
props.put("session.timeout.ms", 45000);
props.put("heartbeat.interval.ms", 15000);
props.put("partition.assignment.strategy", 
    "org.apache.kafka.clients.consumer.StickyAssignor");

Kafka提供三种分区分配策略：

RangeAssignor（默认）：按分区序号范围分配。将分区按数值排序后均分给消费者。Topic级别独立分配，多Topic场景下前排消费者总是分到更多分区，造成不均衡。

RoundRobinAssignor：将所有Topic的所有分区展开为列表，轮询分配给消费者。多Topic场景下均衡性更好，但仍可能出现分区跨Topic分配不合理的情况。

StickyAssignor（推荐）：粘性分配。初次分配追求均衡，重平衡时尽量保持原有分配不变，只迁移必要分区。减少重平衡时的分区迁移量，降低消费恢复时间。

// 自定义RebalanceListener处理重平衡事件
consumer.subscribe(Collections.singletonList("order-events"), 
    new ConsumerRebalanceListener() {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            // 分区被撤销前：提交已处理消息的offset
            for (TopicPartition tp : partitions) {
                long offset = consumer.position(tp);
                commitOffset(tp, offset);
                log.info("分区撤销: {} 当前offset: {}", tp, offset);
            }
        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            // 分区分配后：初始化处理状态
            for (TopicPartition tp : partitions) {
                long committed = getCommittedOffset(tp);
                if (committed >= 0) {
                    consumer.seek(tp, committed);
                }
                log.info("分区分配: {} 从offset: {} 开始", tp, committed);
            }
        }
    });

StickyAssignor粘性分配与重平衡优化

StickyAssignor的核心优势在于重平衡时最小化分区迁移。假设6个分区3个消费者，初始分配为C1[P0,P1], C2[P2,P3], C3[P4,P5]。当C2下线时：

// RangeAssignor重平衡结果（重新排序分配）
// C1[P0,P1,P2], C3[P3,P4,P5]
// P2从C2迁移到C1，P3从C2迁移到C3 -- 2个分区迁移

// StickyAssignor重平衡结果（保持已有分配）
// C1[P0,P1,P2], C3[P4,P5,P3]
// P2迁移到C1，P3迁移到C3 -- C1和C3的原有分区不变
// 消费者无需重新建立分区状态，恢复更快

// CooperativeStickyAssignor（Kafka 2.4+增量式重平衡）
// 配置：
// props.put("partition.assignment.strategy",
//     "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");

// 传统Eager重平衡：所有消费者撤销全部分区 -> 重新分配
// Cooperative重平衡：只撤销需要迁移的分区，其他分区继续消费
// 大幅减少Stop-the-World时间

增量式协同重平衡（Cooperative Rebalance）是Kafka 2.4引入的改进。传统重平衡采用”全部撤销再重新分配”的Eager协议，重平衡期间所有消费者暂停。Cooperative协议分两轮完成：第一轮只撤销需要迁移的分区，第二轮将撤销的分区分配给目标消费者。未迁移分区的消费者可继续消费，显著减少暂停时间。

消费者Offset管理与Exactly-Once语义

Offset管理是保证消息不丢不重的关键。自动提交（enable.auto.commit=true）在后台定时提交poll返回的最新offset，存在消息处理失败但offset已提交的风险。生产环境推荐手动提交。

// 手动同步提交：处理完成后提交
while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

    for (ConsumerRecord<String, String> record : records) {
        try {
            // 业务处理
            processOrder(record.value());

            // 处理成功后提交该消息的offset
            consumer.commitSync(Collections.singletonMap(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
            ));
        } catch (Exception e) {
            log.error("处理失败，不提交offset，下次poll重新消费", e);
            break;
        }
    }
}

// 批量提交：处理完一批后统一提交
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
for (ConsumerRecord<String, String> record : records) {
    processOrder(record.value());
    offsets.put(
        new TopicPartition(record.topic(), record.partition()),
        new OffsetAndMetadata(record.offset() + 1)
    );
}
if (!offsets.isEmpty()) {
    consumer.commitAsync(offsets, (committed, exception) -> {
        if (exception != null) {
            log.error("异步提交失败", exception);
        }
    });
}

// 关闭时同步提交确保最后一批offset不丢
try {
    consumer.commitSync();
} finally {
    consumer.close();
}

Exactly-Once语义需要消费者端的事务性处理配合。将消息消费和业务写入放在同一数据库事务中，或使用Kafka事务API将消费offset提交和下游生产放在同一事务中。消息中间件的可靠性不仅依赖Kafka自身的容错机制，更依赖消费端的正确offset管理策略。在高并发场景下，合理配置poll参数（max.poll.records、max.poll.interval.ms）避免消费者被误判为僵尸进程而触发不必要的重平衡。