第4章 Kafka核心原理与集群部署

Kafka 在实时数仓里扮演什么角色？说白了，它就是整个数据流的「中枢神经」。所有业务数据、日志、埋点，都得先经过 Kafka 做缓冲和解耦，下游的 Flink、Spark 才能放心消费。我做了这么多年实时数仓，还没见过哪个生产环境敢不用 Kafka 的。

这一章，咱们就把 Kafka 的核心原理掰开揉碎，再手把手搭一套集群出来。

4.1 Kafka 架构：一张图看懂全局

先看整体架构。Kafka 集群由多个 Broker 组成，每个 Broker 就是一台服务器。Producer 发消息，Consumer 拉消息，ZooKeeper 负责协调元数据。

Kafka 核心架构图 Producer 生产者集群 Kafka Broker 集群 Broker 1 Broker 2 Broker 3 ZooKeeper 元数据协调 Consumer 消费者集群 Topic / Partition Partition 0 Partition 1 Partition 2 说明：Producer 将消息写入 Topic 的 Partition，Consumer 从 Partition 拉取消息。 ZooKeeper 管理 Broker 元数据和 Leader 选举。

这张图我建议你多看两眼。Producer 不直接找 Consumer，而是把数据扔给 Broker。Broker 负责存，Consumer 自己来取。这种「推拉分离」的设计，让生产者和消费者完全解耦。

我的经验： 刚接触 Kafka 时，我总以为 Producer 会「推」给 Consumer。其实 Kafka 是典型的拉模式——Consumer 主动去 Broker 拉数据。这样做的好处是 Consumer 可以按自己的速度消费，不会把下游打爆。

4.2 Topic 与 Partition：数据分片的核心

Topic 是逻辑上的消息分类。比如订单数据放 order_topic，用户行为放 user_behavior_topic。但真正存数据的是 Partition。

每个 Partition 是一个有序的、不可变的日志文件。消息追加到尾部，Consumer 通过偏移量（Offset）来定位读取位置。

4.2.1 Partition 的分布策略

假设一个 Topic 有 3 个 Partition，集群有 3 个 Broker。Kafka 会尽量把 Partition 分散到不同 Broker 上。这样做的好处很明显：

  • 负载均衡：不会让一台机器扛所有读写
  • 高可用：每个 Partition 可以设置副本，副本分布在其他 Broker 上
  • 并行度：一个 Partition 只能被一个 Consumer 消费，Partition 越多，消费并行度越高

关键点： Partition 的数量决定了消费的并行上限。我见过有人把 Partition 设成 1，然后抱怨消费速度上不去——这不废话吗？

4.2.2 副本机制

每个 Partition 可以有多个副本。副本分为 Leader 和 Follower。Producer 只写 Leader，Follower 从 Leader 同步数据。如果 Leader 挂了，Follower 会选出一个新的 Leader。

参数 说明 推荐值
replication.factor 副本数量 生产环境至少 2，建议 3
min.insync.replicas 最小同步副本数 配合 acks=all 使用，建议设为 2
acks Producer 确认级别 0（不等待）/ 1（Leader 确认）/ all（所有副本确认）
避坑指南： 我曾经把 replication.factor 设为 2，但 min.insync.replicas 也设为 2。结果一台 Broker 宕机后，所有写操作都失败了。因为只剩 1 个副本在线，达不到最小同步数。后来我把 min.insync.replicas 改成 1，配合 acks=all，既保证了数据不丢，又容忍了单点故障。

4.3 生产者与消费者：消息的来与去

4.3.1 生产者核心参数

写 Producer 代码不难，难的是配好参数。我整理了一份常用配置清单：

// Producer 核心配置
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.1.10:9092,192.168.1.11: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");                    // 等待所有副本确认
props.put("retries", 3);                     // 重试次数
props.put("enable.idempotence", true);       // 开启幂等性，防止重复

// 吞吐量相关
props.put("batch.size", 16384);              // 批次大小，16KB
props.put("linger.ms", 5);                   // 延迟发送时间，5ms
props.put("compression.type", "snappy");     // 压缩，减少网络开销

为什么 acks=all 还要开幂等性？你想想看，如果网络抖动导致 Producer 重试，Broker 可能收到两条相同的消息。幂等性通过 Producer ID 和序列号去重，保证 Exactly Once 语义。

4.3.2 消费者核心参数

Consumer 的配置更讲究，尤其是消费位移的管理：

// Consumer 核心配置
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.1.10:9092,192.168.1.11:9092");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
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("fetch.max.bytes", 52428800);      // 每次拉取最大字节，50MB
我的习惯： 生产环境我从来不用自动提交位移。自动提交虽然省事，但一旦 Consumer 处理完消息还没提交就挂了，重启后会重复消费。手动提交虽然代码多几行，但心里踏实。

4.4 Kafka 集群搭建：三台机器就够了

理论说完了，咱们动手搭集群。我习惯用三台机器，每台部署一个 Broker。这样既满足高可用，又不会太浪费资源。

4.4.1 环境准备

  • 三台 CentOS 7 服务器，IP 分别为 192.168.1.10、192.168.1.11、192.168.1.12
  • JDK 1.8+（Kafka 依赖 Java 环境）
  • ZooKeeper 3.6+（Kafka 2.8 之前必须依赖 ZK）
  • Kafka 2.13-3.2.0（我选这个版本，稳定且社区活跃）

4.4.2 安装步骤

  1. 下载解压：三台机器都执行
wget https://archive.apache.org/dist/kafka/3.2.0/kafka_2.13-3.2.0.tgz
tar -xzf kafka_2.13-3.2.0.tgz -C /opt/
ln -s /opt/kafka_2.13-3.2.0 /opt/kafka
  1. 配置 server.properties（以 Broker 1 为例）
# /opt/kafka/config/server.properties
broker.id=1
listeners=PLAINTEXT://192.168.1.10:9092
log.dirs=/data/kafka/logs
num.partitions=3
default.replication.factor=2
zookeeper.connect=192.168.1.10:2181,192.168.1.11:2181,192.168.1.12:2181

其他两台机器类似，broker.id 分别设为 2 和 3，listeners 改成对应 IP。

  1. 启动 ZooKeeper（三台都启动）
# 使用 Kafka 自带的 ZK（生产环境建议独立部署）
/opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties
  1. 启动 Kafka
/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties
  1. 验证集群
# 创建测试 Topic
/opt/kafka/bin/kafka-topics.sh --create \
  --bootstrap-server 192.168.1.10:9092 \
  --replication-factor 2 \
  --partitions 3 \
  --topic test_topic

# 查看 Topic 详情
/opt/kafka/bin/kafka-topics.sh --describe \
  --bootstrap-server 192.168.1.10:9092 \
  --topic test_topic

如果看到 Partition 分布在不同的 Broker 上，副本也正常同步，说明集群搭好了。

验证结果示例：

Topic: test_topic  PartitionCount: 3  ReplicationFactor: 2
  Topic: test_topic  Partition: 0  Leader: 1  Replicas: 1,2  Isr: 1,2
  Topic: test_topic  Partition: 1  Leader: 2  Replicas: 2,3  Isr: 2,3
  Topic: test_topic  Partition: 2  Leader: 3  Replicas: 3,1  Isr: 3,1

Leader 分布在三个 Broker 上，每个 Partition 有两个副本，ISR（同步副本）全部在线。完美。

4.4.3 生产环境调优建议

配置项 默认值 生产推荐 说明
num.network.threads 3 CPU 核数 网络处理线程，I/O 密集型可多配
num.io.threads 8 CPU 核数 * 2 磁盘 I/O 线程
log.segment.bytes 1GB 1GB 日志段大小，太小会导致频繁滚动
log.retention.hours 168 72 日志保留时间，根据业务调整
message.max.bytes 1MB 10MB 单条消息最大大小，日志数据通常较大
曾经踩过的坑： 有次我把 log.retention.hours 设为 1，想省磁盘空间。结果凌晨数据还没被下游消费完就被删了，第二天业务方找我要数据，我只能说「抱歉，没了」。后来我改成 72 小时，配合定时清理策略，再没出过问题。

嗯，这一章的内容就到这。Kafka 的原理和部署其实不复杂，但细节很多。你只要把架构图记在脑子里，把参数配置理解透，生产环境基本不会出大问题。

← 上一章 📖 返回目录 下一章 →