Kafka 概念与部署手册

一、Kafka 核心概念

Kafka 是一个分布式、高吞吐量、低延迟的消息队列系统，主要用于实时数据管道和流处理。以下是核心概念：

1. Producer（生产者）向 Kafka 集群发送消息的客户端，可将消息发布到指定的主题（Topic）。

2. Consumer（消费者）从 Kafka 集群读取消息的客户端，通过订阅主题消费消息。

3. Topic（主题）消息的分类标识，所有消息必须属于某个主题。主题是逻辑概念，物理上消息分散存储在多个分区（Partition）中。

4. Partition（分区）主题的物理拆分单元，每个主题可包含多个分区（提高并行性）。分区内消息有序（按偏移量 Offset 排序），但跨分区无序。

5. Broker（代理节点）Kafka 集群中的服务器节点，负责存储消息、处理生产 / 消费请求。一个集群由多个 Broker 组成。

6. Replica（副本）分区的备份，用于容错。每个分区有一个 Leader 副本（处理读写）和多个 Follower 副本（同步 Leader 数据，Leader 故障时替代）。

7. Consumer Group（消费者组）多个消费者组成的组，共同消费一个主题的消息。同组内消费者分工：每个分区的消息仅被组内一个消费者消费（避免重复消费）。

8. Offset（偏移量）分区内消息的唯一序号，消费者通过记录 Offset 确定已消费位置，支持消息回溯。

二.kafka的集群概念

Kafka 的部署架构是一个分布式系统设计，核心目标是实现高可用、高吞吐量和水平扩展。其架构围绕 Broker 集群、元数据管理、数据存储 和 客户端交互 四个核心维度展开

[Producer] → [Broker 1] ←→ [ZooKeeper 集群][Broker 2] ←→ [ZooKeeper 集群] ←→ [Consumer Group]
[Producer] → [Broker 3] ←→ [ZooKeeper 集群]       (Consumer 1, 2, 3)

核心组件包括：

• Producer（生产者）：向集群发送消息的客户端；
• Broker 集群：由多个服务器节点组成，负责存储消息和处理读写请求；
• ZooKeeper 集群：管理 Kafka 元数据（如集群拓扑、分区副本状态等）；
• Consumer Group（消费者组）：由多个消费者组成，共同消费主题消息。

核心架构组件详解

1. Broker 集群（核心存储与处理节点）

• 作用：每个 Broker 是一台独立服务器，负责存储消息、处理生产者的写入请求和消费者的读取请求。
• 分布式设计：
  • 集群中 Broker 节点通过唯一的 broker.id 标识（配置在 server.properties 中）；
  • 节点间无主从关系，通过 ZooKeeper 协同维护集群状态（如 Broker 加入 / 退出、分区副本选举等）；
  • 可通过增加 Broker 节点水平扩展集群容量和吞吐量。

2. 主题（Topic）与分区（Partition）

• 主题（Topic）：消息的逻辑分类，所有消息必须属于某个主题（类似 “消息队列名称”）。
• 分区（Partition）：
  • 主题的物理拆分单元，每个主题可包含多个分区（数量可配置，默认 1 个），分区数越多，并行处理能力越强；
  • 分区是 Kafka 并行读写的最小单位：生产者可向不同分区并行写入，消费者组可通过多个消费者并行读取不同分区；
  • 分区内消息按 偏移量（Offset） 顺序存储（Offset 是分区内消息的唯一序号，从 0 开始递增），但跨分区无序。

3. 副本（Replica）机制（高可用核心）

• 作用：为每个分区创建多个副本，防止单点故障导致数据丢失，保证集群可用性。
• 副本类型：
  • Leader 副本：每个分区有且仅有一个 Leader 副本，负责处理该分区的所有读写请求；
  • Follower 副本：同步 Leader 副本的数据（通过拉取 Leader 的日志），仅在 Leader 故障时通过选举成为新 Leader；
• 副本分配：
  • 副本分散存储在不同 Broker 上（避免单节点故障导致所有副本失效）；
  • 配置 default.replication.factor 定义默认副本数（建议 3，与集群节点数匹配）。

ZooKeeper 集群（元数据管理）

Kafka 依赖 ZooKeeper 存储集群元数据（Kafka 2.8+ 支持无 ZooKeeper 模式，但主流仍用 ZooKeeper），具体职责：

• 存储 Broker 节点列表（跟踪集群中活跃的 Broker）；
• 记录主题的分区信息（分区数量、副本分布、Leader 副本位置等）；
• 管理消费者组的 Offset 信息（旧版本，新版本默认存储在内部主题 __consumer_offsets 中）；
• 协调 Broker 节点的 Leader 选举（当 Leader 副本故障时）。

生产者（Producer）与消费者（Consumer）交互

• 生产者：

  • 发送消息时指定主题，通过 Kafka 客户端的分区策略（默认按消息键哈希，无键则轮询）决定写入哪个分区；
  • 可配置消息确认机制（acks 参数）：acks=0（不确认）、acks=1（仅 Leader 确认）、acks=all（所有副本确认，最高可靠性）。

• 消费者组：

  • 多个消费者组成一个组，共同消费一个主题的消息，组内每个消费者负责消费部分分区（一个分区仅被组内一个消费者消费，避免重复消费）；
  • 消费者通过记录 Offset 跟踪已消费位置，支持消息回溯（重新消费历史消息）；
  • 当消费者加入 / 退出组时，会触发 重平衡（Rebalance），重新分配分区与消费者的映射关系。

三.kafka集群部署

1. 环境准备

• 依赖：Java 8+（Kafka 运行在 JVM 上）
• 服务器：建议至少 3 台服务器组成集群（Broker 节点），示例节点：node1:192.168.1.101、node2:192.168.1.102、node3:192.168.1.103
• 端口：Broker 间通信默认 9092（客户端端口）、ZooKeeper 默认 2181（Kafka 依赖 ZooKeeper 管理元数据）

2. 安装 ZooKeeper（Kafka 元数据管理）

Kafka 依赖 ZooKeeper 存储集群元数据（如 Broker 列表、分区副本信息等），需先部署 ZooKeeper 集群。

下载并解压：

wget https://archive.apache.org/dist/zookeeper/zookeeper-3.8.4/apache-zookeeper-3.8.4-bin.tar.gz
tar -zxvf apache-zookeeper-3.8.4-bin.tar.gz -C /opt/
mv /opt/apache-zookeeper-3.8.4-bin /opt/zookeeper

配置 ZooKeeper：

cd /opt/zookeeper/conf
cp zoo_sample.cfg zoo.cfg  # 复制默认配置
vi zoo.cfgdataDir=/opt/zookeeper/data  # 数据存储目录
clientPort=2181
# 集群节点配置（server.编号=IP:通信端口:选举端口）
server.1=node1:2888:3888
server.2=node2:2888:3888
server.3=node3:2888:3888

设置节点 ID：

mkdir -p /opt/zookeeper/data
# 在 node1 执行
echo 1 > /opt/zookeeper/data/myid
# 在 node2 执行
echo 2 > /opt/zookeeper/data/myid
# 在 node3 执行
echo 3 > /opt/zookeeper/data/myid

启动 ZooKeeper：

/opt/zookeeper/bin/zkServer.sh start  # 各节点分别启动
/opt/zookeeper/bin/zkServer.sh status  # 检查状态（Mode 应为 leader/follower）

3. 安装 Kafka

下载并解压：

wget https://archive.apache.org/dist/kafka/3.6.1/kafka_2.13-3.6.1.tgz
tar -zxvf kafka_2.13-3.6.1.tgz -C /opt/
mv /opt/kafka_2.13-3.6.1 /opt/kafka

配置 Kafka Broker：

cd /opt/kafka/config
vi server.properties

核心配置（每节点需修改 broker.id 和 advertised.listeners）

# 节点唯一 ID（1、2、3 分别对应 node1、node2、node3）
broker.id=1
# 消息存储目录
log.dirs=/opt/kafka/logs
# 监听地址（客户端连接用，需改为当前节点 IP）
advertised.listeners=PLAINTEXT://node1:9092
# ZooKeeper 集群地址
zookeeper.connect=node1:2181,node2:2181,node3:2181
# 分区副本数（建议 3，与集群节点数一致）
default.replication.factor=3
# 每个主题默认分区数
num.partitions=3

启动 Kafka 集群：

# 后台启动（各节点分别执行）
/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties
# 检查进程
jps  # 应看到 Kafka 进程

4.验证 Kafka 集群

• 创建主题（在任意节点执行）：

  /opt/kafka/bin/kafka-topics.sh --create \--topic test-topic \--bootstrap-server node1:9092,node2:9092,node3:9092 \--partitions 3 \--replication-factor 3

查看主题列表：

/opt/kafka/bin/kafka-topics.sh --list \--bootstrap-server node1:9092

发送消息（生产者）：

/opt/kafka/bin/kafka-console-producer.sh \--topic test-topic \--bootstrap-server node1:9092
# 输入消息并回车发送

消费消息（消费者）：

/opt/kafka/bin/kafka-console-consumer.sh \--topic test-topic \--bootstrap-server node1:9092 \--from-beginning  # 从头消费

5. 停止集群

• 先停止 Kafka：

/opt/kafka/bin/kafka-server-stop.sh

      再停止 ZooKeeper：

/opt/zookeeper/bin/zkServer.sh stop

6. Broker 集群的扩展能力

Kafka 支持通过 新增 Broker 节点 水平扩展集群，无需停机，具体扩展流程如下：

1. 新增 Broker 配置：在新节点上安装 Kafka，配置 broker.id（需唯一）、zookeeper.connect（指向现有 ZooKeeper 集群），其他配置与现有节点保持一致（如 log.dirs、default.replication.factor 等）。
2. 启动新 Broker：新节点启动后，自动注册到 ZooKeeper 的 /brokers/ids 路径，现有集群感知到新节点加入。
3. 分区迁移（可选）：通过 Kafka 工具（kafka-reassign-partitions.sh）将部分分区的副本迁移到新 Broker，平衡集群负载。例如：

   # 创建迁移计划（将 test-topic 的部分分区迁移到新 broker.id=4）
   /opt/kafka/bin/kafka-reassign-partitions.sh --generate \--topics-to-move-json-file topics.json \--broker-list "1,2,3,4" \--zookeeper zk1:2181,zk2:2181,zk3:2181

1. 验证扩展结果：通过 kafka-topics.sh 查看分区副本分布，确认新 Broker 已承担部分副本存储或 Leader 角色。

7.Broker 集群的关键配置参数

以下是影响 Broker 集群性能和可用性的核心配置（在 server.properties 中设置）：

配置参数	作用说明	默认值
broker.id	Broker 唯一标识（整数，需唯一）	0
log.dirs	消息存储目录（多个目录用逗号分隔，可分散到不同磁盘）	/tmp/kafka-logs
zookeeper.connect	ZooKeeper 集群地址（格式：host1:port1,host2:port2）	localhost:2181
default.replication.factor	新建主题的默认副本数（建议 3）	1
num.partitions	新建主题的默认分区数（越多并行性越高，需根据集群规模调整）	1
log.retention.hours	消息保留时间（超过则删除）	168（7 天）
log.segment.bytes	单个日志段大小（写满后创建新段）	1073741824（1GB）
min.insync.replicas	消息写入成功所需的最小同步副本数（确保数据可靠性）	1
replica.lag.time.max.ms	Follower 副本允许的最大同步延迟（超过则被踢出 ISR 列表）	30000（30 秒）