(kafka集群搭建)
一、环境准备
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.启动时可能会遇到以下报错： 
这个时候请检查端口是否放开，若没有则有两种方法：
- 关闭防火墙
#sudo systemctl stop firewalld
- 或者防火墙放行
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
#我这里有三台机器所以设置下面的属性为3
offsets.topic.replication.factor=3 transaction.state.log.replication.factor=3 transaction.state.log.min.isr=3
group.initial.rebalance.delay.ms=3
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 测试结果
发布者： 
消费者 
测试成功！