4. Kafka 生产与消费实战：从客户端到生产环境

Kafka 的生产与消费，说白了就是两个核心动作：谁把数据放进去，谁把数据拿出来。但实际落地时，分区策略、消费者组、再均衡、偏移量提交……这些细节才是真正考验人的地方。我刚开始用 Kafka 时，觉得写个 Producer 发消息很简单，结果线上丢数据、重复消费、分区倾斜，踩了个遍。今天咱们就把这些实战要点掰开揉碎。

4.1 消息生产：Java 与 Python 客户端

先看 Java。Kafka 官方对 Java 的支持最成熟，生产环境首选。我习惯用 KafkaProducer 配合 ProducerRecord，配置项里 bootstrap.servers、key.serializer、value.serializer 是必须的。

// Java Producer 示例
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 生产环境建议加 acks=all 保证不丢数据
props.put("acks", "all");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 100; i++) {
    ProducerRecord<String, String> record = 
        new ProducerRecord<>("my-topic", "key-" + i, "value-" + i);
    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            System.err.println("发送失败: " + exception.getMessage());
        }
    });
}
producer.close();

Python 这边，我推荐 confluent-kafka 库，它底层封装了 librdkafka，性能比 kafka-python 好不少。配置方式类似，但用字典传参。

# Python Producer 示例
from confluent_kafka import Producer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'key.serializer': str.encode,
    'value.serializer': str.encode,
    'acks': 'all'
}
producer = Producer(conf)

def delivery_report(err, msg):
    if err is not None:
        print(f'发送失败: {err}')
    else:
        print(f'发送成功: {msg.topic()} [{msg.partition()}]')

for i in range(100):
    producer.produce('my-topic', key=f'key-{i}', value=f'value-{i}',
                     callback=delivery_report)
producer.flush()

4.2 分区策略：数据该去哪

Kafka 的分区策略决定了消息落到哪个 Partition。默认策略是 DefaultPartitioner：如果 key 不为 null，对 key 的哈希值取模；如果 key 为 null，轮询分配。但实际场景里，我们经常需要自定义策略。

举个例子，订单数据按用户 ID 分区，保证同一个用户的所有订单进入同一个分区，这样消费时能保证顺序。我做过一个电商项目，订单处理必须严格按时间顺序，当时就自定义了一个分区器。

// 自定义分区器：按用户 ID 的哈希值分区
public class UserIdPartitioner 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();
        // 假设 key 是用户 ID 字符串
        String userId = (String) key;
        return Math.abs(userId.hashCode()) % numPartitions;
    }
    
    @Override
    public void close() {}
    
    @Override
    public void configure(Map<String, ?> configs) {}
}

Python 里自定义分区器稍微麻烦点，需要继承 Partitioner 类。不过大多数场景下，用默认策略就够了。你想想看，如果业务没有严格的分区顺序要求，轮询反而更均衡。

策略	适用场景	注意事项
默认哈希（key 非 null）	需要按 key 保证顺序	key 分布不均会导致数据倾斜
轮询（key 为 null）	无顺序要求，追求负载均衡	无法保证同一类数据在同一个分区
自定义分区器	业务有特殊路由需求	注意分区数变化时的兼容性

4.3 消费者组与再均衡

消费者组是 Kafka 实现水平扩展和容错的核心机制。同一个 Group ID 下的消费者共同消费一个 Topic，每个分区只能被组内的一个消费者消费。说白了，就是「分工协作，互不抢活」。

再均衡（Rebalance）发生在消费者加入、退出或分区数变化时。Kafka 会重新分配分区，这个过程会导致短暂的消费中断。我见过不少新手在再均衡期间丢数据，就是因为没处理好偏移量提交。

// Java Consumer 示例
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 自动提交偏移量，默认每 5 秒一次
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

try {
    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());
        }
    }
} finally {
    consumer.close();
}

Python 的消费者写法类似，但要注意 poll() 的超时时间设置。

# Python Consumer 示例
from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': True,
    'auto.commit.interval.ms': 5000
}
consumer = Consumer(conf)
consumer.subscribe(['my-topic'])

try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            print(f"Consumer error: {msg.error()}")
            continue
        print(f'offset = {msg.offset()}, key = {msg.key()}, value = {msg.value()}')
finally:
    consumer.close()

4.4 偏移量提交机制

偏移量（Offset）是消费者在分区中的读取位置。提交偏移量就是告诉 Kafka：「我已经处理到这个位置了，下次从这里继续读」。提交方式分两种：自动提交和手动提交。

自动提交简单省事，但容易丢数据或重复消费。比如你设置了 5 秒自动提交，但消息处理了 6 秒，这时候消费者挂了，重启后就会从 5 秒前的偏移量开始读，导致重复消费。反过来，如果消息还没处理完就提交了偏移量，消费者挂了就会丢数据。

手动提交更可控。我建议在「消息处理完成」之后提交偏移量，这样能最大程度保证「至少一次」语义。

// 手动提交偏移量（Java）
props.put("enable.auto.commit", "false");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // 处理消息
            process(record);
        }
        // 同步提交，阻塞直到提交成功
        consumer.commitSync();
    }
} finally {
    consumer.close();
}

Python 里手动提交也很直接：

# 手动提交偏移量（Python）
conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'enable.auto.commit': False
}
consumer = Consumer(conf)
consumer.subscribe(['my-topic'])

try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            continue
        # 处理消息
        process(msg)
        # 手动提交当前偏移量
        consumer.commit(asynchronous=False)
finally:
    consumer.close()

提交方式	优点	缺点	推荐场景
自动提交	代码简单，无需手动管理	可能丢数据或重复消费	允许少量重复/丢失的非关键数据
手动同步提交	精确控制，可靠性高	阻塞，影响吞吐量	对数据一致性要求高的场景
手动异步提交	非阻塞，吞吐量高	提交失败时可能丢失偏移量	高吞吐场景，配合回调处理失败

4.5 知识体系总览

下面这张图把本章的核心知识点串起来了。从生产者到消费者，分区策略、消费者组、再均衡、偏移量提交，每一步都有坑，也都有解法。

嗯，到这里 Kafka 生产与消费的核心实战内容就讲完了。从客户端代码到分区策略，从消费者组到偏移量提交，每一步都有讲究。记住一点：生产环境里，宁可慢一点，也要稳一点。自动提交虽然方便，但手动提交才是企业级应用的正道。