一、Kafka 基础与系统要求

1.1 核心概念

Broker：Kafka 服务器节点，负责存储消息和处理客户端请求

Topic：消息分类的逻辑容器，每条消息需指定发送到某个 Topic

Partition：Topic 的物理分片，可分布式存储在不同 Broker 上，实现水平扩展

Producer：消息生产者，向 Kafka Topic 发送消息

Consumer：消息消费者，从 Kafka Topic 订阅并消费消息

Consumer Group：多个 Consumer 组成的分组，同组内 Consumer 共同消费 Topic 消息（避免重复消费）

ZooKeeper：Kafka 早期版本（2.8 前）依赖的分布式协调工具，用于管理集群元数据、选举 Broker 控制器等（2.8+ 版本支持无 ZooKeeper 模式）

1.2 系统要求

基础环境：Java 8 或 Java 11（推荐，Kafka 依赖 JVM 运行）

硬件配置：

- 单节点：至少 2GB 内存、20GB 磁盘空间（测试环境）

- 集群节点：建议 4GB+ 内存、100GB+ 磁盘（生产环境，优先 SSD 提升性能）

支持系统：Linux（Ubuntu/CentOS，推荐生产环境）、Windows（仅建议测试环境）、macOS

网络要求：集群节点间网络互通，开放默认端口（Broker 9092、ZooKeeper 2181）

二、Kafka 安装（单节点，测试环境）

2.1 依赖环境准备（ZooKeeper 安装）

Kafka 2.8 版本前需单独部署 ZooKeeper，以下以 Kafka 2.13-3.5.1 版本（依赖 ZooKeeper） 为例：

2.1.1 Linux 系统（Ubuntu/CentOS 通用）

安装流程图解：

执行步骤：

1. 下载 ZooKeeper（推荐 3.8.x 版本）：

wget https://dlcdn.apache.org/zookeeper/zookeeper-3.8.4/apache-zookeeper-3.8.4-bin.tar.gz

2. 解压到 /opt 目录：

tar -zxvf apache-zookeeper-3.8.4-bin.tar.gz -C /opt/
ln -s /opt/apache-zookeeper-3.8.4-bin /opt/zookeeper  # 创建软链接，方便后续操作

3. 修改配置文件：

cd /opt/zookeeper/conf
cp zoo_sample.cfg zoo.cfg  # 复制默认配置为正式配置
# 编辑配置，指定数据存储目录
sed -i 's/dataDir=\/tmp\/zookeeper/dataDir=\/opt/zookeeper/data/' zoo.cfg

4. 创建数据目录并启动服务：

mkdir -p /opt/zookeeper/data
/opt/zookeeper/bin/zkServer.sh start  # 启动服务

5. 验证状态（出现 Mode: standalone 即为单节点运行成功）：

/opt/zookeeper/bin/zkServer.sh status

2.1.2 Windows 系统

安装流程图解：

执行步骤：

1. 下载 ZooKeeper 压缩包（同 Linux 步骤 1），解压到 D:\zookeeper

2. 进入 D:\zookeeper\conf，复制 zoo_sample.cfg 并重命名为 zoo.cfg，修改数据目录：

dataDir=D:\zookeeper\data  # 手动创建 data 文件夹

3. 配置环境变量：

- 系统变量新增 ZOOKEEPER_HOME，值为 D:\zookeeper

- 在 Path 变量中添加 %ZOOKEEPER_HOME%\bin

4. 启动服务（以管理员身份打开 cmd）：

zkServer.cmd  # 启动服务，保持窗口运行（关闭窗口则服务停止）

5. 验证服务：新打开 cmd 窗口，执行 zkCli.cmd，若能进入 [zk: localhost:2181(CONNECTED)] 则成功。

2.2 Kafka 安装（Linux 系统为例）

2.2.1 安装流程图解

2.2.2 执行步骤

1. 下载 Kafka（选择与 Scala 兼容的版本，如 2.13-3.5.1）：

wget https://dlcdn.apache.org/kafka/3.5.1/kafka_2.13-3.5.1.tgz

2. 解压到 /opt 目录并创建软链接：

tar -zxvf kafka_2.13-3.5.1.tgz -C /opt/
ln -s /opt/kafka_2.13-3.5.1 /opt/kafka

3. 修改核心配置文件 server.properties（关键配置项）：

cd /opt/kafka/config
vi server.properties

主要修改内容：

broker.id=0  #  Broker 唯一标识，集群中需不同（如 0、1、2）
listeners=PLAINTEXT://:9092  # 监听地址，默认 9092 端口（集群需指定节点 IP）
log.dirs=/opt/kafka/logs  # 消息日志存储目录（手动创建 logs 文件夹）
zookeeper.connect=localhost:2181  # ZooKeeper 地址（集群需填写所有 ZooKeeper 节点，如 zk1:2181,zk2:2181）
auto.create.topics.enable=false  # 禁用自动创建 Topic，建议手动创建

4. 启动 Kafka Broker 服务（需先确保 ZooKeeper 已启动）：

# 前台启动（用于测试，关闭终端则服务停止）
/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties
# 后台启动（生产环境推荐，日志输出到 logs/kafkaServer.out）
/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

5. 验证 Broker 状态（查看 9092 端口是否被占用）：

netstat -tuln | grep 9092  # 出现 LISTEN 状态即为成功

2.3 Windows 系统 Kafka 安装

与 Linux 步骤类似，核心差异：

1. 下载 Kafka 压缩包并解压到 D:\kafka

2. 修改 D:\kafka\config\server.properties：

log.dirs=D:\kafka\logs  # 手动创建 logs 文件夹
zookeeper.connect=localhost:2181

3. 启动 Broker（以管理员身份打开 cmd，进入 D:\kafka\bin\windows）：

kafka-server-start.bat ../../config/server.properties  # 前台启动

4. 验证：查看 9092 端口是否占用（cmd 执行 netstat -ano | findstr "9092"）。

三、Kafka 核心配置（单节点与集群）

3.1 单节点关键配置（server.properties）

配置项	说明	推荐值（测试环境）
broker.id	Broker 唯一 ID（集群中不可重复）	0
listeners	监听地址格式：协议://IP:端口，PLAINTEXT 为明文协议	PLAINTEXT://:9092
log.dirs	消息日志存储目录（建议独立磁盘，避免与系统盘共用）	/opt/kafka/logs
zookeeper.connect	ZooKeeper 连接地址（集群格式：zk1:2181,zk2:2181,zk3:2181/kafka）	localhost:2181
log.retention.hours	消息保留时间（超过时间自动删除）	168（7 天）
log.segment.bytes	单个日志片段大小（达到阈值后滚动生成新片段）	1073741824（1GB）
num.partitions	新建 Topic 的默认分区数（分区越多，吞吐越高，但资源消耗也越大）	3

3.2 Kafka 集群部署（3 节点示例）

3.2.1 集群架构图解

3.2.2 集群部署步骤（基于 Linux 系统）

准备 3 台服务器，分别命名为 kafka1（192.168.1.101）、kafka2（192.168.1.102）、kafka3（192.168.1.103），确保节点间网络互通，且已安装 Java 环境。
部署 ZooKeeper 集群（3 节点）：

- 每台节点按 2.1.1 步骤安装 ZooKeeper，修改 zoo.cfg 新增集群配置：

# 集群节点配置：server.节点ID=IP:通信端口:选举端口
server.1=192.168.1.101:2888:3888
server.2=192.168.1.102:2888:3888
server.3=192.168.1.103:2888:3888

- 每台节点在 dataDir 目录下创建 myid 文件，写入对应节点 ID（如 kafka1 写 1，kafka2 写 2）：

echo 1 > /opt/zookeeper/data/myid  # kafka1 节点执行

- 依次启动 3 台节点的 ZooKeeper 服务，验证集群状态（其中一台为 Leader，其余为 Follower）：

/opt/zookeeper/bin/zkServer.sh status

3. 部署 Kafka 集群：

- 每台节点按 2.2 步骤安装 Kafka，修改 server.properties 关键配置：

节点	broker.id	listeners	zookeeper.connect
kafka1	0	PLAINTEXT://192.168.1.101:9092	192.168.1.101:2181,192.168.1.102:2181,192.168.1.103:2181
kafka2	1	PLAINTEXT://192.168.1.102:9092	同上
kafka3	2	PLAINTEXT://192.168.1.103:9092	同上

- 依次启动 3 台节点的 Kafka Broker 服务（后台启动）：

/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

- 验证集群状态（查看所有 Broker 是否在线）：

# 列出集群中所有 Broker 节点
/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.1.101:9092

四、Kafka 基本操作（命令行工具）

4.1 Topic 管理

Kafka 提供命令行工具用于 Topic 的创建、查询、修改和删除，工具位于 /opt/kafka/bin 目录（Windows 系统在 D:\kafka\bin\windows）。

4.1.1 创建 Topic

命令格式（Linux 系统）：

# 创建 Topic（指定分区数 3、副本数 2，需集群环境支持副本）
/opt/kafka/bin/kafka-topics.sh \
--bootstrap-server 192.168.1.101:9092,192.168.1.102:9092 \
--create \
--topic user-behavior-topic \
--partitions 3 \
--replication-factor 2 \
--config retention.ms=604800000  # 消息保留时间（7天，可选）

参数说明：

--bootstrap-server：指定 Kafka 集群节点（单节点用 localhost:9092）

--partitions：Topic 分区数（建议与 Broker 数量匹配，提升吞吐）

--replication-factor：副本数（需 ≤ Broker 数量，确保数据冗余，生产环境建议 2-3）

Windows 系统命令：

kafka-topics.bat --bootstrap-server localhost:9092 --create --topic user-behavior-topic --partitions 3 --replication-factor 1

4.1.2 查询 Topic 列表与详情

# 查询所有 Topic
/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list# 查询指定 Topic 详情（含分区、副本分布）
/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic user-behavior-topic

4.1.3 修改 Topic 配置（如分区数、保留时间）

# 增加分区数（注意：分区数仅能增加，不能减少）
/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic user-behavior-topic --partitions 5# 修改消息保留时间（改为 3 天）
/opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --topic user-behavior-topic --add-config retention.ms=259200000

4.1.4 删除 Topic

/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic user-behavior-topic

4.2 生产者与消费者测试（命令行）

4.2.1 启动生产者发送消息

# Linux 系统
/opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic user-behavior-topic# Windows 系统
kafka-console-producer.bat --bootstrap-server localhost:9092 --topic user-behavior-topic

启动后直接输入消息（如 {"user_id": "1001", "behavior": "click", "timestamp": 1690000000}），按回车发送。

4.2.2 启动消费者接收消息

新打开终端，执行以下命令（消费者需指定消费组，默认组为 console-consumer-xxx）：

# 从头消费 Topic 所有消息（--from-beginning 可选，默认消费新消息）
/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic user-behavior-topic --from-beginning --group test-consumer-group

此时生产者发送的消息会实时在消费者终端显示，验证消息收发正常。

五、Kafka 实际应用场景（代码示例）

5.1 场景说明：用户行为日志采集

假设需要采集用户在电商平台的点击、下单、支付等行为日志，通过 Kafka 实现 “日志采集→消息存储→实时消费” 流程，架构如下：

5.2 Java 生产者代码示例（Spring Boot 集成）

5.2.1 引入依赖（pom.xml）

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.9.2</version> <!-- 与 Spring Boot 版本匹配 -->
</dependency>

5.2.2 配置 Kafka 生产者（application.yml）

spring:kafka:bootstrap-servers: 192.168.1.101:9092,192.168.1.102:9092producer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.springframework.kafka.support.serializer.JsonSerializeracks: 1  # 消息确认机制（1：Leader 接收成功即确认，平衡性能与可靠性）retries: 3  # 发送失败重试次数batch-size: 16384  # 批量发送大小（16KB）linger-ms: 5  # 等待 5ms 批量发送（提升效率）

5.2.3 生产者代码（发送用户行为日志）

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;@Component
public class UserBehaviorProducer {private final KafkaTemplate<String, UserBehaviorLog> kafkaTemplate;private static final String TOPIC = "user-behavior-topic";// 构造注入 KafkaTemplatepublic UserBehaviorProducer(KafkaTemplate<String, UserBehaviorLog> kafkaTemplate) {this.kafkaTemplate = kafkaTemplate;}// 发送消息（key 为用户 ID，用于分区路由，确保同一用户消息进入同一分区）public void sendUserBehaviorLog(UserBehaviorLog log) {kafkaTemplate.send(TOPIC, log.getUserId(), log).addCallback(success -> System.out.println("消息发送成功：" + log),failure -> System.err.println("消息发送失败：" + failure.getMessage()));}
}// 用户行为日志实体类
class UserBehaviorLog {private String userId;private String behavior; // click, order, payprivate Long timestamp;// Getter、Setter 省略
}

5.3 Java 消费者代码示例（监听 Topic）

5.3.1 配置 Kafka 消费者（application.yml）

spring:kafka:consumer:group-id: user-behavior-consumer-group  # 消费组 IDkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.springframework.kafka.support.serializer.JsonDeserializerauto-offset-reset: earliest  # 无偏移量时从头消费（可选：latest 消费新消息）enable-auto-commit: false  # 禁用自动提交偏移量，手动提交确保消息不丢失listener:ack-mode: manual-immediate  # 手动立即提交偏移量

5.3.2 消费者监听代码

import org.apache.kafka.clients.consumer.ConsumerAcknowledgment;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;@Component
public class UserBehaviorConsumer {// 监听指定 Topic，手动提交偏移量@KafkaListener(topics = "user-behavior-topic", groupId = "user-behavior-consumer-group")public void consumeUserBehavior(ConsumerRecord<String, UserBehaviorLog> record, ConsumerAcknowledgment ack) {try {// 1. 获取消息内容String userId = record.key();UserBehaviorLog log = record.value();System.out.println("接收用户行为日志：" + userId + " - " + log.getBehavior());// 2. 业务处理（如存储到 Elasticsearch、触发实时计算）// elasticsearchTemplate.save(log);// 3. 手动提交偏移量（确保业务处理完成后提交，避免消息丢失）ack.acknowledge();} catch (Exception e) {System.err.println("消息处理失败：" + e.getMessage());// 失败时可重试或死信队列处理}}
}

六、Kafka 维护与优化

6.1 日志清理与磁盘管理

Kafka 消息默认存储在 log.dirs 目录，需定期监控磁盘空间，避免磁盘占满导致服务异常：

6.1.1 配置日志清理策略（server.properties）

# 日志清理策略：delete（删除过期日志）或 compact（按 key 压缩，保留最新值）
log.cleanup.policy=delete# 单个日志片段大小（达到后滚动生成新片段，默认 1GB）
log.segment.bytes=1073741824# 日志片段保留时间（超过后触发清理，默认 7 天）
log.retention.hours=168# 日志目录磁盘使用率阈值（超过 85% 触发清理，默认 85%）
log.dirs.config=log.retention.bytes=10737418240  # 可选：按磁盘大小限制（10GB）

6.1.2 手动触发日志清理（紧急情况）

# 查看 Topic 日志目录
ls -l /opt/kafka/logs/user-behavior-topic-*# 手动删除过期日志片段（需谨慎，建议先备份）
rm -rf /opt/kafka/logs/user-behavior-topic-0/00000000000000000000.log.1690000000

6.2 性能优化建议

6.2.1 Broker 优化（server.properties）

1. 内存配置：修改 Kafka 启动脚本 bin/kafka-server-start.sh，调整 JVM 内存（生产环境建议 4-8GB）：

export KAFKA_HEAP_OPTS="-Xms4g -Xmx4g"  # 堆内存大小（避免超过物理内存 50%）

2. 网络优化：

# 每个 Broker 最大并发连接数（默认 2000）
num.network.threads=8
# 处理磁盘 I/O 的线程数（默认 8，建议与 CPU 核心数匹配）
num.io.threads=16
# 发送缓冲区大小（默认 128KB，高吞吐场景可调整为 256KB）
socket.send.buffer.bytes=262144
# 接收缓冲区大小（默认 128KB，同上）
socket.receive.buffer.bytes=262144

6.2.2 生产者优化（application.yml）

spring:kafka:producer:batch-size: 32768  # 批量发送大小（32KB，提升吞吐）linger-ms: 10  # 等待 10ms 批量发送（平衡延迟与吞吐）compression-type: gzip  # 消息压缩（gzip/snappy，减少网络传输）acks: 1  # 仅 Leader 确认（生产环境建议 1，兼顾可靠性与性能）

6.2.3 消费者优化

spring:kafka:consumer:fetch.min.bytes: 16384  # 最小拉取字节数（16KB，避免频繁拉取）fetch.max.wait.ms: 500  # 最大等待时间（500ms，超时即使未达阈值也拉取）listener:concurrency: 3  # 消费者并发数（建议等于 Topic 分区数，提升消费速度）

6.3 数据备份与灾难恢复

6.3.1 定期备份 Topic 数据

# 使用 kafka-dump-log 工具导出 Topic 数据到文件（示例：导出分区 0 的数据）
/opt/kafka/bin/kafka-dump-log.sh --files /opt/kafka/logs/user-behavior-topic-0/00000000000000000000.log --print-data-log > user-behavior-backup.log

6.3.2 集群故障恢复（如 Broker 下线）

单个 Broker 下线：若配置了副本（replication-factor ≥2），下线 Broker 上的分区会自动切换到其他副本，无需手动干预，重启下线 Broker 后会自动同步数据并加入集群。
数据丢失恢复：若未配置副本导致数据丢失，需从备份文件恢复，步骤如下：

# 1. 创建临时 Topic
/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic temp-topic --partitions 3 --replication-factor 1# 2. 使用生产者导入备份数据
/opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic temp-topic < user-behavior-backup.log# 3. 将临时 Topic 数据迁移到原 Topic（需开发脚本或使用 Flink/Spark 同步）

七、常见故障排查

7.1 Broker 启动失败

故障现象：执行启动命令后，服务很快退出，查看日志报错。

排查步骤：

1. 查看日志：日志路径默认在 logs/server.log，执行：

tail -f /opt/kafka/logs/server.log

2. 常见错误与解决：

- 错误 1：Address already in use（端口被占用）

解决：查看 9092 端口占用进程，杀死进程或修改 server.properties 中的 listeners 端口：

netstat -tuln | grep 9092  # 查看占用进程
kill -9 <进程ID>  # 杀死进程

- 错误 2：Connection refused to ZooKeeper（无法连接 ZooKeeper）

解决：检查 ZooKeeper 服务是否启动，或 zookeeper.connect 配置是否正确：

/opt/zookeeper/bin/zkServer.sh status  # 检查 ZooKeeper 状态

7.2 生产者发送消息失败

故障现象：生产者发送消息后报错 Leader not available。

排查步骤：

检查 Topic 是否存在，执行 kafka-topics.sh --list 确认。
检查集群 Broker 状态，执行 kafka-broker-api-versions.sh 确认所有 Broker 在线。
若为集群环境，检查 listeners 配置是否为节点 IP（而非 localhost），否则跨节点无法访问。

7.3 消费者无法接收消息

故障现象：消费者启动后，无消息接收，生产者发送正常。

排查步骤：

检查消费组 group-id 是否正确，不同消费组可重复消费同一 Topic。
检查 auto-offset-reset 配置：若为 latest，仅消费启动后的新消息；改为 earliest 可从头消费。
检查 Topic 分区与消费组偏移量，执行以下命令查看偏移量：

/opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group user-behavior-consumer-group

若 Current offset 等于 Log end offset，说明已消费完所有消息，需生产者发送新消息。

八、总结

Kafka 作为高性能的分布式消息队列，核心优势在于高吞吐、低延迟和高可靠性，广泛应用于日志采集、实时计算、消息通知等场景。通过本文的学习，读者可掌握：

单节点与集群环境的安装部署（含 ZooKeeper 配置）；
Topic、生产者、消费者的核心操作；
实际应用场景的代码集成（Spring Boot 示例）；
日常维护、性能优化与故障排查方法。

在生产环境中，需根据业务流量合理规划分区数、副本数，结合监控工具（如 Prometheus + Grafana）实时监控集群状态，确保 Kafka

一、Kafka 基础与系统要求

1.1 核心概念

1.2 系统要求

二、Kafka 安装（单节点，测试环境）

2.1 依赖环境准备（ZooKeeper 安装）

2.1.1 Linux 系统（Ubuntu/CentOS 通用）

2.1.2 Windows 系统

2.2 Kafka 安装（Linux 系统为例）

2.2.1 安装流程图解

2.2.2 执行步骤

2.3 Windows 系统 Kafka 安装

三、Kafka 核心配置（单节点与集群）

3.1 单节点关键配置（server.properties）

3.2 Kafka 集群部署（3 节点示例）

3.2.1 集群架构图解

3.2.2 集群部署步骤（基于 Linux 系统）

四、Kafka 基本操作（命令行工具）

4.1 Topic 管理

4.1.1 创建 Topic

4.1.2 查询 Topic 列表与详情

4.1.3 修改 Topic 配置（如分区数、保留时间）

4.1.4 删除 Topic

4.2 生产者与消费者测试（命令行）

4.2.1 启动生产者发送消息

4.2.2 启动消费者接收消息

五、Kafka 实际应用场景（代码示例）

5.1 场景说明：用户行为日志采集

5.2 Java 生产者代码示例（Spring Boot 集成）

5.2.1 引入依赖（pom.xml）

5.2.2 配置 Kafka 生产者（application.yml）

5.2.3 生产者代码（发送用户行为日志）

5.3 Java 消费者代码示例（监听 Topic）

5.3.1 配置 Kafka 消费者（application.yml）

5.3.2 消费者监听代码

六、Kafka 维护与优化

6.1 日志清理与磁盘管理

6.1.1 配置日志清理策略（server.properties）

6.1.2 手动触发日志清理（紧急情况）

6.2 性能优化建议

6.2.1 Broker 优化（server.properties）

6.2.2 生产者优化（application.yml）

6.2.3 消费者优化

6.3 数据备份与灾难恢复

6.3.1 定期备份 Topic 数据

6.3.2 集群故障恢复（如 Broker 下线）

七、常见故障排查

7.1 Broker 启动失败

故障现象：执行启动命令后，服务很快退出，查看日志报错。

7.2 生产者发送消息失败

故障现象：生产者发送消息后报错 Leader not available。

7.3 消费者无法接收消息

故障现象：消费者启动后，无消息接收，生产者发送正常。

八、总结

相关文章