Henry's xyz

外观

Kafka Consumer ​

消费模型 ​

Kafka 采用 Pull 模型：Consumer 主动从 Broker 拉取消息，按自身处理能力控制消费速率，避免 Push 模式下的消费方过载。

Consumer Group "order-service"
  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐
  │ Consumer-1   │  │ Consumer-2   │  │ Consumer-3   │
  │ Partition-0   │  │ Partition-1   │  │ Partition-2   │
  └──────────────┘  └──────────────┘  └──────────────┘
        ▲                  ▲                  ▲
        └──────────────────┼──────────────────┘
                     Topic "orders" (3 Partitions)

同一 Group 内，一个 Partition 只能被一个 Consumer 消费；Consumer 数 > Partition 数时，多余的 Consumer 闲置。

基本用法 ​

java
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "order-service");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders", "payments"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset=%d, key=%s, value=%s%n",
            record.offset(), record.key(), record.value());
    }
}
shell
# 命令行消费（从头开始）
kafka-console-consumer.sh --topic orders --from-beginning \
  --bootstrap-server localhost:9092

# 指定 Consumer Group
kafka-console-consumer.sh --topic orders --group my-group \
  --bootstrap-server localhost:9092

Offset 管理 ​

模式说明适用
自动提交enable.auto.commit=true，定期自动提交 Offset简单场景，容忍少量重复
手动同步提交consumer.commitSync()必须确认不丢失
手动异步提交consumer.commitAsync()高吞吐场景
java
// 自动提交（默认）
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000");

// 手动同步提交 — 每次 poll 后立即提交
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        process(record);  // 先处理，再提交
    }
    consumer.commitSync();  // 阻塞直到提交成功
}

// 手动异步提交 — 高吞吐
consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        log.error("提交失败", exception);
    }
});

// 按 Partition 精确提交 Offset
consumer.commitSync(Map.of(
    new TopicPartition("orders", 0), new OffsetAndMetadata(12345)
));
场景推荐策略
日志收集、指标上报（可容忍丢失）自动提交
订单处理（不可丢失，少量重复可接受）手动同步提交
高吞吐、可容忍少量重复手动异步 + consumer.close() 前同步
不可重复（需 Exactly-Once）配合幂等生产者或事务

Rebalance ​

Consumer Group 内成员增减、Topic Partition 增减时触发 Rebalance，重新分配 Partition。

分区分配策略 ​

策略配置值行为
RangeRangeAssignor按 Topic 范围分（默认），可能不均衡
Round-RobinRoundRobinAssignor轮询分配所有 Partition
StickyStickyAssignor尽量维持原有分配，减少数据迁移
Cooperative StickyCooperativeStickyAssignorSticky + 渐进式再均衡（推荐，2.4+）
java
// 推荐：渐进式 Rebalance，不触发 stop-the-world
props.put("partition.assignment.strategy",
    "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");

// 3.1+ 默认策略
props.put("group.protocol", "consumer");  // 新的 Consumer Group Protocol

Rebalance 钩子 ​

java
consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 当前分配被撤销前，同步提交 Offset
        consumer.commitSync();
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 新分配到达，按需 seek 到特定 Offset
        for (TopicPartition tp : partitions) {
            consumer.seek(tp, getLastCheckpointedOffset(tp));
        }
    }
});

核心配置 ​

参数默认值说明
group.id—Consumer Group 标识
enable.auto.committrue自动提交 Offset
auto.offset.resetlatest无 Offset 时的行为：latest / earliest / none
max.poll.records500单次 poll 最大消息数
max.poll.interval.ms300000 (5min)两次 poll 的最大间隔，超时触发 Rebalance
session.timeout.ms45000心跳超时，超时踢出 Group
fetch.min.bytes1最少拉取字节数（调大可提升吞吐）
fetch.max.wait.ms500最大等待时间（配合 fetch.min.bytes）
isolation.levelread_uncommittedread_committed 可过滤未提交的事务消息

参考 ​