🐇明明跟你说过：个人主页
🏅个人专栏：《大数据前沿：技术与应用并进》🏅
🔖行路有良友，便是天堂🔖
目录
一、引言
1、Kafka简介
2、Kafka核心优势
二、环境准备
1、服务器
2、服务器环境初始化
三、安装zookeeper
1、上传tar包
2、编辑配置文件
3、创建数据目录
4、安装JAVA
5、启动zookeeper
四、Kafka集群搭建
1、上传tar包
2、编辑配置文件
3、创建数据目录
4、启动kafka
5、查看端口状态
五、测试
1、创建Topic
2、生产消息
3、消费消息
一、引言
1、Kafka简介
Apache Kafka 是一个开源的分布式流处理平台，它最初由 LinkedIn 开发，后来成为 Apache 软件基金会的顶级项目。Kafka 设计用于处理实时数据流，提供了一种高效、可扩展、持久化的方式来进行数据发布和订阅。它通常被描述为一种分布式发布-订阅消息队列，但它实际上超越了传统消息队列的概念。
2、Kafka核心优势
1. 高吞吐量：
- Kafka 能够处理海量数据，支持每秒数十万条消息的读写操作，即使在大规模部署中也能保持高性能。
- 通过高效的文件系统设计和内存管理机制，Kafka 能够在处理大量数据的同时保持低延迟。
2. 持久性和可靠性：
- Kafka 将数据存储在磁盘上，并支持数据复制（replication），确保即使在节点故障的情况下也能保证数据的可靠性和持久性。
- 数据以追加的方式写入日志文件，减少了磁盘的随机写操作，提高了写入速度和数据完整性。
3. 可扩展性：
- Kafka 具有良好的水平扩展能力，可以通过增加更多的节点来提升系统的处理能力和存储容量。
- 分布式架构使得 Kafka 能够轻松地在多台服务器上部署，并且能够动态扩展和收缩集群大小。
4. 灵活的发布-订阅模型：
- Kafka 支持发布-订阅模式，允许多个消费者订阅同一个主题，并且消费者可以独立消费消息。
- 消费者可以控制自己的消费进度，不会影响其他消费者的状态，实现了消息消费的解耦。
二、环境准备
1、服务器
准备3台或者5台Linux服务器，用来组建高可用集群，这里使用3台Centos 7.9来进行搭建，大家也可以使用其他的Linux发行版本
配置如下：
2、服务器环境初始化
3台机器都要执行
关闭Selinux
vi /etc/selinux/config#修改成如下
SELINUX=disabled之后重启服务器
reboot关闭并禁用防火墙
[root@kafka1 ~]# systemctl stop firewalld && systemctl disable firewalld修改 /etc/hosts
vi /etc/hosts# 添加以下内容
192.168.40.100 kafka1
192.168.40.101 kafka2
192.168.40.102 kafka3修改镜像源
curl -o /etc/yum.repos.d/CentOS-Base.repo https://mirrors.aliyun.com/repo/Centos-7.repo三、安装zookeeper
为什么安装Kafka时，要先安装zookeeper：
ZooKeeper 是一个分布式的协调服务，它为分布式应用程序提供了一套完整的协调服务功能，包括命名服务、配置管理、集群管理和同步等。Kafka 利用 ZooKeeper 来管理其集群中的多个组件，确保系统的稳定性和一致性。
1、上传tar包
apache-zookeeper-3.8.0-bin.tar.gz
tar包可以去官网进行下载 Apache ZooKeeper
https://zookeeper.apache.org/releases.html#download
解压tar包至 /opt 下
tar zxvf apache-zookeeper-3.8.0-bin.tar.gz -C /opt2、编辑配置文件
vim /opt/apache-zookeeper-3.8.0-bin/conf/zoo.cfg# 输入如下内容
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/opt/zookeeper/zkData
dataLogDir=/opt/zookeeper/zkLog
clientPort=2181
server.1=kafka1:2188:3888
server.2=kafka2:2188:3888
server.3=kafka3:2188:3888
4lw.commands.whitelist=*
tickTime=2000：
- tickTime 定义了 ZooKeeper 服务器之间的心跳间隔时间（毫秒）。它是 ZooKeeper 中最基本的单位时间。默认值通常是 2000 毫秒（即 2 秒）。
initLimit=10：
- initLimit 定义了初始同步阶段的最大超时时间（心跳次数）。这意味着在初始同步阶段，跟随者（follower）必须在 initLimit * tickTime 毫秒内完成与领导者（leader）的同步。例如，这里设置为 20 秒（10 * 2000 毫秒）。
syncLimit=5：
- syncLimit 定义了在领导者和跟随者之间发送消息的最大超时时间（心跳次数）。这意味着在同步阶段，跟随者必须在 syncLimit * tickTime 毫秒内响应领导者的请求。例如，这里设置为 10 秒（5 * 2000 毫秒）。
dataDir=/opt/zookeeper/zkData：
- dataDir 指定 ZooKeeper 服务器用来存储快照（snapshot）的目录。
dataLogDir=/opt/zookeeper/zkLog：
- dataLogDir 指定 ZooKeeper 服务器用来存储事务日志（transaction logs）的目录。这是从 ZooKeeper 3.4.6 开始引入的一个配置项，使得日志和数据可以分开存储。
clientPort=2181：
- clientPort 指定客户端连接到 ZooKeeper 服务器的端口，默认为 2181。
server.1=ka1:2188:3888：
- server.N 表示第 N 台服务器的信息，格式为 hostname:peerPort:leaderPort。peerPort 是服务器之间通信的端口，leaderPort 是选举领导者时使用的端口。
4lw.commands.whitelist=*：
- 4lw.commands.whitelist 指定客户端可以执行的命令白名单。* 表示允许所有命令。
3、创建数据目录
mkdir -p /opt/zookeeper/zkData
mkdir -p /opt/zookeeper/zkLog创建集群ID文件
在3台机器上分别执行
[root@kafka1 bin]# echo 1 > /opt/zookeeper/zkData/myid
[root@kafka2 bin]# echo 2 > /opt/zookeeper/zkData/myid
[root@kafka3 bin]# echo 3 > /opt/zookeeper/zkData/myid4、安装JAVA
yum install -y java-1.8.0-openjdk-devel5、启动zookeeper
cd /opt/apache-zookeeper-3.8.0-bin/bin/
./zkServer.sh start ../conf/zoo.cfg &
查看状态
[root@kafka1 bin]# ps -aux | grep zook
四、Kafka集群搭建
1、上传tar包
资源包大家可以到官网下载
https://kafka.apache.org/
解压至指定目录
[root@kafka1 ~]# tar zxvf kafka_2.13-3.1.0.tgz -C /opt/
[root@kafka2 ~]# tar zxvf kafka_2.13-3.1.0.tgz -C /opt/
[root@kafka3 ~]# tar zxvf kafka_2.13-3.1.0.tgz -C /opt/2、编辑配置文件
在kafka1上执行
[root@kafka1 config]# vim /opt/kafka_2.13-3.1.0/config/server.properties#输入如下内容
broker.id=0
listeners=PLAINTEXT://kafka1:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/kafka-logs
num.partitions=5
default.replication.factor=2
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181/kafka
zookeeper.connection.timeout.ms=18000
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0在kafka2上执行
[root@kafka2 config]# vim /opt/kafka_2.13-3.1.0/config/server.properties#输入如下内容
broker.id=1
listeners=PLAINTEXT://kafka2:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/kafka-logs
num.partitions=5
default.replication.factor=2
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181/kafka
zookeeper.connection.timeout.ms=18000
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0在kafka3上执行
[root@kafka3 config]# vim /opt/kafka_2.13-3.1.0/config/server.properties#输入如下内容
broker.id=2
listeners=PLAINTEXT://kafka3:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/kafka-logs
num.partitions=5
default.replication.factor=2
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181/kafka
zookeeper.connection.timeout.ms=18000
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0
broker.id=0
- 这是 Kafka broker 的唯一标识符。每个 broker 必须有唯一的 ID。这里的值为 0，意味着这是一个集群中的单个 broker 或者是第一个 broker。
listeners=PLAINTEXT://kafka1:9092
- 定义了 broker 监听的网络接口和端口。此处使用 PLAINTEXT 协议，意味着没有加密。kafka1:9092 表示监听名为 kafka1 的主机上的 9092 端口。
num.network.threads=3
- 指定了用于网络请求处理的线程数。网络请求包括接收来自生产者的消息、发送消息给消费者等操作。这里设置为 3 个线程。
num.io.threads=8
- 指定了用于处理 I/O 请求的线程数。I/O 请求包括磁盘上的读写操作。这里设置为 8 个线程。
socket.send.buffer.bytes=102400
- 设置了发送套接字的缓冲区大小（单位：字节）。此配置影响网络数据包的发送速度。此处设置为 102400 字节。
socket.receive.buffer.bytes=102400
- 设置了接收套接字的缓冲区大小（单位：字节）。此配置影响网络数据包的接收速度。此处设置为 102400 字节。
socket.request.max.bytes=104857600
- 定义了从客户端接收的最大请求大小（单位：字节）。这有助于防止因过大请求而导致的内存溢出。此处设置为 104857600 字节，即约 100MB。
log.dirs=/opt/kafka-logs
- 指定了日志文件存储的位置。日志文件包含了 Kafka topic 的数据。这里设置的日志目录为 /opt/kafka-logs。
num.partitions=5
- 指定了默认主题分区的数量。分区越多，通常意味着更高的并发度。这里设置的主题默认分区数为 5。
default.replication.factor=2
- 指定了创建新主题时的默认复制因子。复制因子决定了每个分区的副本数量。这里设置的复制因子为 2，意味着每个分区有 2 份副本。
num.recovery.threads.per.data.dir=1
- 指定了用于恢复日志段的线程数。每个数据目录可以有不同的线程数。这里设置为 1 个线程。
offsets.topic.replication.factor=1
- 指定了 _consumer_offsets 主题的复制因子。此主题用于存储消费者的偏移量信息。这里设置的复制因子为 1，意味着只有一个副本。
transaction.state.log.replication.factor=1
- 指定了 _transactions 主题的复制因子。此主题用于记录事务状态。这里设置的复制因子为 1，意味着只有一个副本。
transaction.state.log.min.isr=1
- 指定了 _transactions 主题的最小 ISR（In-Sync Replicas）数量。ISR 是与 leader 同步的副本集合。这里设置的最小 ISR 数量为 1。
log.retention.hours=168
- 指定了日志数据保留的时间长度（单位：小时）。这里设置的日志保留时间为 168 小时，即 7 天。
log.segment.bytes=1073741824
- 指定了日志段的最大大小（单位：字节）。一旦达到这个大小，Kafka 就会创建一个新的日志段。这里设置的日志段大小为 1073741824 字节，即 1GB。
log.retention.check.interval.ms=300000
- 指定了检查日志清理的间隔时间（单位：毫秒）。这里设置的检查间隔为 300000 毫秒，即 5 分钟。
zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181/kafka
- 指定了 ZooKeeper 服务器列表。ZooKeeper 用于协调 Kafka 集群。这里设置的 ZooKeeper 服务器为 kafka1:2181, kafka2:2181, kafka3:2181，路径为 /kafka。
zookeeper.connection.timeout.ms=18000
- 指定了 Kafka 与 ZooKeeper 之间的连接超时时间（单位：毫秒）。这里设置的超时时间为 18000 毫秒，即 18 秒。请注意，zookeeper.connection.timeout.ms 在配置中出现了两次，应该是误写，只需要保留一次即可。
group.initial.rebalance.delay.ms=0
- 指定了消费者组初始重新平衡的延迟时间（单位：毫秒）。这里设置的延迟时间为 0，即立即开始重新平衡。
3、创建数据目录
在3台机器上分别执行
mkdir /opt/kafka-logs4、启动kafka
[root@kafka1 bin]# cd /opt/kafka_2.13-3.1.0/bin/
[root@kafka1 bin]# ./kafka-server-start.sh -daemon ../config/server.properties[root@kafka2 bin]# cd /opt/kafka_2.13-3.1.0/bin/
[root@kafka2 bin]# ./kafka-server-start.sh -daemon ../config/server.properties[root@kafka3 bin]# cd /opt/kafka_2.13-3.1.0/bin/
[root@kafka3 bin]# ./kafka-server-start.sh -daemon ../config/server.properties5、查看端口状态
[root@kafka1 bin]# netstat -antupl
五、测试
1、创建Topic
前面我们已经将kafka集群搭建起来了，接下来创建一个Topic进行写入测试，如果不清楚Topic是什么，可以翻看作者之前的文章。
在kafka1上执行
[root@kafka1 bin]# cd /opt/kafka_2.13-3.1.0/bin
[root@kafka1 bin]# ./kafka-topics.sh --bootstrap-server=192.168.40.100:9092 --topic test --create --partitions=3 --replication-factor=2- --bootstrap-server：指定 Kafka broker 的地址和端口号。这里的 192.168.40.100:9092 指定了 broker 的 IP 地址为 192.168.40.100，端口号为 9092。
- --topic：指定要操作的主题名称。在这个例子中，主题名为 test。
- --create：告诉 Kafka 创建一个新主题。如果主题已经存在，这条命令将会失败，除非你配置了允许创建已存在的主题。
- --partitions：指定主题的分区数。分区数决定了主题能够并行处理消息的能力。在这个例子中，主题 test 将会有 3 个分区。
- --replication-factor：指定主题的复制因子。复制因子决定了每个分区的副本数量，这对于数据的冗余和可靠性非常重要。在这个例子中，主题 test 的每个分区将会有 2 个副本。
查看Topic
[root@kafka1 bin]# ./kafka-topics.sh --bootstrap-server=192.168.40.100:9092 --list
2、生产消息
[root@kafka1 bin]# ./kafka-console-producer.sh --bootstrap-server 192.168.40.100:9092 --topic test向我们刚刚创建的test Topic写入几条消息
3、消费消息
[root@kafka1 bin]# ./kafka-console-consumer.sh --bootstrap-server=192.168.40.100:9092 --topic test --from-beginning
如果能看到之前生产的消息，则证明集群搭建成功
💕💕💕每一次的分享都是一次成长的旅程，感谢您的陪伴和关注。希望这些关于大数据的文章能陪伴您走过技术的一段旅程，共同见证成长和进步！😺😺😺
🧨🧨🧨让我们一起在技术的海洋中探索前行，共同书写美好的未来！！！