📚 课程目录第4章 Kafka消息队列实战

4. Kafka消息队列实战：Kafka核心概念与实时数据管道搭建

好，咱们进入第四章。这一章我打算跟你好好聊聊Kafka——实时数据管道里最核心的组件。说实话，我在做用户事件归因系统之前，对消息队列的理解也就停留在“能发能收”的层面。直到有一次线上事故，消费者组配置错了，数据重复消费了整整两小时……嗯，从那以后我才真正把Kafka的核心概念吃透。

这一章咱们不讲虚的，直接上手。从核心概念到安装配置，再到API使用和可靠性保证，一条线走下来。

4.1 Kafka核心概念：Topic、Partition、Consumer Group

先说说这三个最基础也最容易搞混的概念。

4.1.1 Topic（主题）

Topic说白了就是数据分类的标签。你往Kafka里写数据，必须指定一个Topic。比如用户点击事件发到 user_click，支付事件发到 user_payment。每个Topic可以理解成一个逻辑上的数据流管道。

我的习惯：Topic命名尽量带上业务域和版本号，比如 user_click_v2。别问我为什么——有一次重构时发现十几个Topic名字乱成一锅粥，改都不敢改。

4.1.2 Partition（分区）

Partition是Kafka实现高吞吐的关键。一个Topic可以拆成多个Partition，每个Partition是一个有序的、不可变的日志文件。数据写入时按Partition分布，消费时也可以并行读取。

我举个例子：假设 user_click 有3个Partition，你往里面写100万条数据，Kafka会按key的哈希值把数据分散到不同Partition。这样3个消费者可以同时读，速度直接翻三倍。

概念	类比	关键点
Topic	数据库中的表	逻辑分类，不存数据
Partition	表的分区文件	物理存储，有序不可变
Offset	文件中的行号	唯一标识每条消息位置

4.1.3 Consumer Group（消费者组）

Consumer Group是Kafka实现“一条消息只被消费一次”的机制。同一个Group内的消费者共同消费一个Topic，每条消息只会被其中一个消费者处理。不同Group之间互不干扰。

你想想看：如果两个服务都需要消费 user_click 数据——一个做实时归因，一个做离线分析——它们应该属于不同的Group。这样两个服务都能收到全量数据，互不影响。

我曾经踩过的坑：消费者组内消费者数量不能超过Topic的Partition数。超过的消费者会一直空闲，不报错也不消费。排查了半天才发现是Partition不够分。

4.2 Kafka安装与配置

咱们直接实战。我习惯用Docker快速搭建Kafka环境，但也会讲一下手动安装的要点。

4.2.1 使用Docker Compose快速部署

这是我个人最推荐的方式。写一个 docker-compose.yml，一条命令启动所有服务。

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

启动命令：

docker-compose up -d

等几秒钟，Kafka就起来了。你可以用 docker ps 确认两个容器都在运行。

4.2.2 手动安装（Linux环境）

如果你不想用Docker，手动安装也很简单：

下载Kafka二进制包：wget https://downloads.apache.org/kafka/3.5.0/kafka_2.13-3.5.0.tgz
解压：tar -xzf kafka_2.13-3.5.0.tgz
启动ZooKeeper：bin/zookeeper-server-start.sh config/zookeeper.properties
启动Kafka：bin/kafka-server-start.sh config/server.properties

避坑指南：手动安装时记得修改 server.properties 里的 advertised.listeners，否则客户端连不上。我当初就是忘了改这个，折腾了半小时。

4.3 生产者与消费者API使用

好，环境搭好了，咱们写代码。我用Java API做演示，因为Kafka原生API就是Java写的，最稳定。

4.3.1 生产者API

生产者负责把数据发到Kafka。核心参数就几个：

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");
props.put("acks", "all");  // 可靠性关键参数

Producer<String, String> producer = new KafkaProducer<>(props);

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

这里有个关键参数 acks。我解释一下：

acks值	含义	可靠性	性能
0	不等待确认	最低	最高
1	Leader确认即可	中等	中等
all	所有副本确认	最高	最低

做实时归因时，我建议用 acks=all。数据丢了可不是闹着玩的。

4.3.2 消费者API

消费者负责从Kafka拉数据。核心代码：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "realtime-attribution-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");  // 手动提交offset

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

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());
        // 处理业务逻辑
    }
    consumer.commitSync();  // 处理完再提交
}

  重点：我习惯把 enable.auto.commit 设为 false，手动提交offset。这样能保证“处理完再提交”，避免数据丢失。自动提交虽然方便，但一旦程序崩溃，没处理完的数据就丢了。

4.4 消息可靠性保证

说到可靠性，这是实时归因系统的生命线。我总结了几条铁律：

4.4.1 生产者端：幂等性与事务

Kafka从0.11版本开始支持幂等生产者。开启后，Kafka会自动去重，避免网络重试导致的数据重复。

props.put("enable.idempotence", "true");
props.put("transactional.id", "tx-user-click-001");  // 事务ID

如果你需要“精确一次”语义，还得配合事务API：

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(record1);
    producer.send(record2);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

4.4.2 消费者端：手动提交与重试

消费者端最容易出问题。我见过太多人因为自动提交导致数据丢失。记住这个原则：先处理，后提交。

try {
    // 处理业务逻辑
    processRecords(records);
    // 处理成功才提交
    consumer.commitSync();
} catch (Exception e) {
    // 记录失败消息，后续重试
    log.error("处理失败，offset: {}", records.iterator().next().offset());
    // 不要提交，下次poll会重新消费
}

4.4.3 副本机制与ISR

Kafka通过副本保证数据不丢。每个Partition可以有多个副本，其中一个Leader，其余是Follower。ISR（In-Sync Replicas）是跟Leader保持同步的副本列表。

配置参数：

# server.properties
default.replication.factor=3
min.insync.replicas=2

这样设置后，至少2个副本同步成功才算写入完成。我建议生产环境至少3副本，min.insync.replicas设为2。

我曾经犯过的错：有一次把 min.insync.replicas 设成了3，结果一台机器宕机后所有写入都失败了。后来改成2，既保证了可靠性，又容忍单点故障。

4.5 本章知识体系总览

下面这张图把Kafka的核心逻辑串起来了。从生产者写入，到Topic/Partition存储，再到消费者组消费，一条线走通。

这张图把咱们今天讲的内容全串起来了。生产者写数据到Topic，Topic按Partition分散存储，消费者组内的消费者并行消费。每个环节都有可靠性保障——生产者用幂等性和事务，消费者用手动提交，存储层用副本和ISR。