Kafka消息队列+zookeeper

前言

在分布式系统架构中，消息队列（Message Queue，MQ） 早已成为解耦、削峰、异步处理的核心组件。当业务无需立即获取结果，但需要控制并发、应对流量峰值、保证数据可靠传递时，消息队列就是最优解。

而 Kafka 作为 Apache 旗下顶级开源项目，凭借高吞吐量、低延迟、分布式、持久化、易扩展等特性，成为大数据领域、微服务架构中最主流的消息队列与流处理平台，广泛应用于日志收集、实时数据同步、流式计算、事件驱动架构等场景。

一、消息队列（MQ）

1. 什么是消息队列？

• 消息（Message）：应用间传递的数据单元，可简单（文本）可复杂（对象）。
• 消息队列（Message Queue）：应用间异步通信框架，发送者只管发、接收者只管取，双方无需感知对方存在，由队列保证消息可靠存储与传递。

2. 消息队列三大核心特征

3. 为什么要用消息队列？（8 大核心优势）

二、Kafka 核心原理与基础概念

1. Kafka 官方定义

Kafka 是高吞吐量的分布式发布 / 订阅消息系统，支持实时处理海量数据，适配 Hadoop 分析、Storm/Spark 流式计算、低时延实时系统。

2. Kafka 核心术语

3. Kafka 拓扑架构

典型 Kafka 集群 = 若干 Producer + 若干 Broker + 若干 Consumer Group + ZooKeeper 集群

• ZooKeeper：负责集群配置、Leader 选举、消费组重平衡。
• Producer：Push（推） 消息到 Broker。
• Consumer：Pull（拉） 消息，自主控制消费速率。
• 核心规则：一个分区只允许同一消费组内一个消费者消费，保证顺序性。

4. Topic 与 Partition 深度解析

• 一个 Topic 可拆分为多个 Partition，分散存储在不同 Broker。
• 好处：突破单机磁盘限制、提升吞吐、负载均衡。
• 物理存储：每个 Partition 对应一个文件夹，包含 .index 索引文件 + .log 数据文件。
• 建议：Partition 数量 ≥ 消费者数量，且 ≤ Broker 数量。

5. Producer 生产机制

生产者根据分区策略将消息分发到不同 Partition，实现负载均衡，并行写入提升吞吐。

6. Consumer 消费机制

• 两种模式：
• 队列模式：一个消费组，一条消息仅被一个消费者消费。
• 发布 / 订阅模式：多个消费组，一条消息可被多组消费。
• 优势：Pull 模式支持消费者批量拉取、自主控速。

三、ZooKeeper 核心原理（Kafka 依赖组件）

1. ZooKeeper 是什么？

ZooKeeper 是分布式协调服务，解决分布式系统中多进程同步、共享资源竞争、脑裂（双主节点）问题，提供分布式锁、集群管理、配置同步等能力。

2. 核心作用

• 解决单点故障：避免传统主备切换导致的双 Master 问题。
• 集群选举：Master 故障自动切换，恢复后自动转为备用节点。

3. ZooKeeper 集群角色

4. ZooKeeper 写数据流程

四、ZooKeeper 在 Kafka 中的 7 大核心作用

1. Broker 注册

• 路径：/brokers/ids
• Broker 启动创建临时节点，记录 IP / 端口，宕机自动删除。

2. Topic 注册

• 路径：/brokers/topics
• 维护 Topic 与 Partition、Broker 的对应关系。

3. 生产者负载均衡

• 动态感知 Broker 上下线，实现消息均匀分发。

4. 消费者负载均衡

• 消费组内消费者合理分配分区，避免重复消费 / 消息丢失。

5. 记录分区与消费者关系

• 路径：/consumers/[group_id]/owners/[topic]/[partition]
• 保证一个分区仅被同组一个消费者消费。

6. 记录消费进度 Offset

• 路径：/consumers/[group_id]/offsets/[topic]/[partition]
• 消费者重启后从断点继续消费。

7. 消费者注册与监听

• 消费者启动注册节点，监听 Broker / 消费组变化，触发负载均衡。

五、Kafka 单机部署

1. 基础环境配置

bash

运行

# 修改主机名
hostnamectl set-hostname kafka1
# 添加映射
echo '192.168.10.101 kafka1' >> /etc/hosts

2. 安装 JDK（ZooKeeper 依赖）

bash

运行

yum -y install java

3. 安装并配置 ZooKeeper

bash

运行

# 解压
tar zxvf apache-zookeeper-3.6.0-bin.tar.gz
mv apache-zookeeper-3.6.0-bin /etc/zookeeper
# 复制配置文件
cd /etc/zookeeper/conf
mv zoo_sample.cfg zoo.cfg
# 创建数据目录
mkdir -p /etc/zookeeper/zookeeper-data

配置 zoo.cfg

plaintext

dataDir=/etc/zookeeper/zookeeper-data
clientPort=2181

bash

运行

# 启动 ZooKeeper
cd /etc/zookeeper
./bin/zkServer.sh start
# 查看状态
./bin/zkServer.sh status

4. 安装并配置 Kafka

bash

运行

# 解压
tar zxvf kafka_2.13-2.4.1.tgz
mv kafka_2.13-2.4.1 /etc/kafka
# 创建日志目录
mkdir -p /etc/kafka/kafka-logs

修改 server.properties

plaintext

broker.id=0
listeners=PLAINTEXT://192.168.10.101:9092
log.dirs=/etc/kafka/kafka-logs
zookeeper.connect=192.168.10.101:2181

bash

运行

# 后台启动 Kafka
nohup ./bin/kafka-server-start.sh config/server.properties &
# 验证端口
netstat -anpt | grep 2181
netstat -anpt | grep 9092

5. Kafka 单机测试

bash

运行

# 修复脚本警告
sed -i 's/egrep/grep -E/' bin/kafka-run-class.sh

# 创建 Topic
bin/kafka-topics.sh –create –zookeeper kafka1:2181 –replication-factor 1 –partitions 1 –topic test

# 查看 Topic 列表
bin/kafka-topics.sh –list –zookeeper kafka1:2181

# 查看 Topic 详情
bin/kafka-topics.sh –describe –zookeeper kafka1:2181 –topic test

# 生产消息
bin/kafka-console-producer.sh –broker-list kafka1:9092 –topic test

# 消费消息（新终端）
bin/kafka-console-consumer.sh –bootstrap-server kafka1:9092 –topic test –from-beginning

# 删除 Topic
bin/kafka-topics.sh –delete –zookeeper kafka1:2181 –topic test

六、Kafka 集群部署（3 节点）

1. 集群资源规划

2. 基础环境（所有节点执行）

bash

运行

# 关闭防火墙
systemctl stop firewalld
setenforce 0

# 添加主机映射
cat >> /etc/hosts << EOF
192.168.10.101 kafka1
192.168.10.102 kafka2
192.168.10.103 kafka3
EOF

3. ZooKeeper 集群配置

plaintext

dataDir=/etc/zookeeper/zookeeper-data
clientPort=2181
server.1=192.168.10.101:2888:3888
server.2=192.168.10.102:2888:3888
server.3=192.168.10.103:2888:3888

bash

运行

# kafka1
echo '1' > /etc/zookeeper/zookeeper-data/myid
# kafka2
echo '2' > /etc/zookeeper/zookeeper-data/myid
# kafka3
echo '3' > /etc/zookeeper/zookeeper-data/myid

bash

运行

./bin/zkServer.sh start
./bin/zkServer.sh status

4. Kafka 集群配置

修改 server.properties（每个节点差异化）

• kafka1：

plaintext

broker.id=1
listeners=PLAINTEXT://192.168.10.101:9092
log.dirs=/etc/kafka/kafka-logs
zookeeper.connect=192.168.10.101:2181,192.168.10.102:2181,192.168.10.103:2181

• kafka2：broker.id=2，IP 改为 192.168.10.102
• kafka3：broker.id=3，IP 改为 192.168.10.103

bash

运行

# 启动 Kafka
nohup ./bin/kafka-server-start.sh config/server.properties &

5. 集群验证

bash

运行

# 任意节点创建 Topic（副本数 2，分区 3）
bin/kafka-topics.sh –create –zookeeper kafka1:2181 –replication-factor 2 –partitions 3 –topic cluster-test

# 所有节点查看 Topic
bin/kafka-topics.sh –list –zookeeper kafka1:2181

# 生产/消费测试
bin/kafka-console-producer.sh –broker-list kafka1:9092,kafka2:9092,kafka3:9092 –topic cluster-test
bin/kafka-console-consumer.sh –bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 –topic cluster-test –from-beginning

七、补充知识点

1. Kafka 高吞吐核心原因

• 顺序写磁盘：避免随机 IO，机械盘也能高性能。
• 页缓存：利用操作系统缓存，减少磁盘 IO。
• 批量发送 / 压缩：Producer 批量发送，支持 Gzip/Snappy 压缩。
• 零拷贝：减少内核态与用户态数据拷贝。

2. Kafka 高可用机制

• 副本机制：每个 Partition 设多个副本，分布在不同 Broker。
• Leader 故障：ZooKeeper 自动从 Follower 选举新 Leader。
• ISR 机制：保持同步的副本集合，仅 ISR 内节点可被选举为 Leader。

3. 消费组重平衡（Rebalance）

触发条件：

• 消费者上线 / 下线
• Topic 分区增减
• Broker 上下线作用：重新分配分区与消费者对应关系，保证负载均衡。

4. Kafka 消息丢失 / 重复消费解决方案

• 消息丢失：设置 acks=all（所有副本确认写入）、开启重试、手动提交 Offset。
• 重复消费：业务端做幂等性（如唯一 ID 去重）。

5. Kafka 常用运维命令

bash

运行

# 修改分区数
bin/kafka-topics.sh –alter –zookeeper kafka1:2181 –topic test –partitions 5

# 查看消费组列表
bin/kafka-consumer-groups.sh –bootstrap-server kafka1:9092 –list

# 查看消费组偏移量
bin/kafka-consumer-groups.sh –bootstrap-server kafka1:9092 –describe –group test-group