一、环境准备

1.1 服务器准备

搭建kafka集群至少需要3台服务器（或虚拟机也可），我们提前准备好3台不同IP的服务器

ip	hostname
192.168.2.83	kafkaserver1
192.168.2.84	kafkaserver2
192.168.2.86	kafkaserver3

1.2 版本信息

kafak：2.13-3.4.0；
JDK：1.8+；

1.3 启动kafka环境

1.3.1 获取kafka

首先下载最新版kafka，再解压安装：

$ tar -xzf kafka_2.13-3.4.0.tgz
$ cd kafka_2.13-3.4.0

完成后可以得到一个kafka的文件夹，我这里是直接放在桌面的，各位可以根据自己需求选择文件夹存放。

1.3.2 启动kafka

Apache Kafka 可以使用 ZooKeeper 或 KRaft 启动，但不能同时使用。

1.3.2.1 kafka && ZooKeeper

以前的kafka之前还需要准备zookeeper环境，新版的kafka中自带由，免去了很多麻烦。

第一步：按顺序启动所有服务：

# Start the ZooKeeper service
$ bin/zookeeper-server-start.sh config/zookeeper.properties

第二步：打开另一个终端会话并运行：

# Start the Kafka broker service
$ bin/kafka-server-start.sh config/server.properties

kafka的bin目录带有zookeeper：

以上两步命令运行后，可以看到kafka服务已经启动：

1.3.2.2 kafak && KRaft

第一步：生成集群 UUID

$ KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

第二步：格式化日志

$ bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties

第三步：启动卡夫卡服务器

$ bin/kafka-server-start.sh config/kraft/server.properties

以上三步命令运行后，同理可以看到kafka服务已经启动，这里就不演示了。

1.4 创建主题

kafka的事件储存在主题中，主题类似于文件系统中的文件夹，事件是该文件夹中的文件。所以，在使用前必须为事件创建主题。

打开另一个终端会话并运行

$ bin/kafka-topics.sh --create --topic kafka-topic-test --bootstrap-server localhost:9092

Kafka 的所有命令行工具都有额外的选项：运行kafka-topics.sh不带任何参数的命令以显示使用信息。例如，显示 新主题的 分区计数等详细信息：

$ bin/kafka-topics.sh --describe --topic kafka-topic-test --bootstrap-server localhost:9092

1.5 事件写入主题

Kafka 客户端通过网络与 Kafka 代理通信以写入（或读取）事件。一旦收到，代理将以持久和容错的方式存储事件。

默认情况下，输入的每一行都会导致一个单独的事件被写入主题。

$ bin/kafka-console-producer.sh --topic kafka-topic-test --bootstrap-server localhost:9092
one
two

1.6 事件读取

打开另一个终端会话并运行控制台消费者客户端以读取事件：

$ bin/kafka-console-consumer.sh --topic kafka-topic-test --from-beginning --bootstrap-server localhost:9092

1.7 停止kafka环境

使用 停止生产者和消费者客户Ctrl-C。 使用 停止 Kafka 代理Ctrl-C。如果使用 Kafka && ZooKeeper，还需要停止 ZooKeeper 服务器 Ctrl-C。

如果还想删除本地 Kafka 环境的任何数据，包括在此过程中创建的任何事件，请运行以下命令

$ rm -rf /tmp/kafka-logs /tmp/zookeeper /tmp/kraft-combined-logs

二、搭建 kafka 集群

前提条件：按照第一步骤的方法为列出的三台服务器全部安装 kakfa 。
我们这里用的 kafka && ZooKeeper 模式，所以先要搭建 zookeeper 集群。

2.1 ZooKeeper 集群搭建

2.1.1 配置文件修改

我们前面已经安装好了单机版的 kafak，它自带 ZooKeeper，现在我们对它们进行配置搭建集群。

cd kafka_2.13-3.4.0/
cd config/
vim zookeeper.properties

修改zookeeper.properties文件：

1.修改文件目录

dataDir=/var/zookeeper
#添加
initLimit=5
syncLimit=2
tickTime=2000

2.添加服务节点

server.1=192.168.2.83:2888:3888
server.2=192.168.2.84:2888:3888
server.3=192.168.2.86:2888:3888

像下面这样：

3.创建数据文件夹

# 新建zookeeper.properties 中配置的dataDir目录
mkdir /var/zookeeper
cd /var/zookeeper
# 新建myid文件,三个节点分别设置值为1,2,3
touch myid
echo 1 > /var/zookeeper/myid

4.启动时可能会遇到以下报错：

这个时候请检查端口是否放开，若没有则有两种方法：

1. 关闭防火墙

#sudo systemctl stop firewalld

2. 或者防火墙放行

firewall-cmd --zone=public --add-port=2181/tcp --permanent 
firewall-cmd --zone=public --add-port=3888/tcp --permanent 
firewall-cmd --zone=public --add-port=2888/tcp --permanent
firewall-cmd --reload

然后在其他的2个服务器重重复上面的步骤，（注意节点设置）

2.2 kafka 集群搭建

2.2.1 kafka 配置修改

1.修改server.properties

#每个节点唯一
broker.id=1
#填写本地IP
listeners=PLAINTEXT://192.168.2.83:9092
#这里有几个成员就写几个
zookeeper.connect=192.168.2.83:2181,192.168.2.84:2181,192.168.2.86:2181

2.启动 kafka

3.如果启动报以下错误

则对meta.properties进行修改，该文件默认在 /tmp/kafka-logs

修改其cluster.id等于Cluster ID即可。

2.2.1 kafka 集群测试

搭建完成后，我们简单的测试一下是否成功。

我们在节点 192.168.2.73上启动一个发布者进行消息发布，在节点192.168.2.84和192.168.2.86上分别启动一个消费者进行事件的消费。

2.2.1.1 创建测试用主题

我们首先创建一个主题 topic-test，并指定分区为 0，读取模式为从头读取，并连接本地服务器。

bin/kafka-console-consumer.sh --bootstrap-server 192.168.2.83:9092 --offset latest --partition 0 --topic topic-test

2.2.1.2 创建消费者

随后我们在每个集群服务其上创建消费者，并指定地址为本地地址。

bin/kafka-console-consumer.sh --bootstrap-server 192.168.2.83:9092 --offset latest --partition 0 --topic topic-test

2.2.1.3 创建发布者

随便选一台服务器创建发布者,指定集群列表和主题，并发布消息：

bin/kafka-console-producer.sh  --topic topic-test --broker-list 192.168.2.83:9092,192.168.2.84:9092,192.168.2.86:9092

2.2.1.4 测试结果

发布者：

消费者

测试成功！