1-Kafka集群搭建
1.安装包解压至相应目录
/home/ZX/soft/kafka_2.12-3.0.0
2.搭建三个节点
IP broker.id myid
10.103.105.91 1 1
10.103.105.92 2 2
10.103.105.93 3 3
3.修改配置文件
（1）zookeeper.properties
dataDir=/home/ZX/soft/kafka_2.12-3.0.0/zookeeper
dataLogDir=/home/ZX/soft/kafka_2.12-3.0.0/datalog
clientPort=2181
maxClientCnxns=100
tickTime=20
initLimit=10
syncLimit=5
server.1=10.103.105.91:2888:3888
server.2=10.103.105.92:2888:3888
server.3=10.103.105.93:2888:3888
三个机器上的zookeeper.properties文件配置相同，需要注意的是日志保存的路径，不会自动生成，需要自己手动建立相关的路径， dataLogDir是自己加的，日志文件太多，把日志文件区分开；
(2)server.properties
server.properties配置文件的修改主要在开头和结尾，中间保持默认配置即可；需要注意的点是broker.id的值三个节点要配置不同的值，分别配置为1，2，3；log.dirs必须保证目录存在，不会根据配置文件自动生成；
注意：以下是要修改地方的代码，并不是只有这几个，其他配置保持默认即可。
broker.id=3
listeners=PLAINTEXT://10.103.105.93:9092
advertised.listeners=PLAINTEXT://10.103.105.93:9092
log.dirs=/home/ZX/soft/kafka_2.12-3.0.0/logs/kafka
zookeeper.connect=10.103.105.91:2181,10.103.105.92:2181,10.103.105.93:2181
三个机器上的server.properties中有所不同；
broker.id，图中两个listenners分别为对应的值。
所以，可在91节点配置好后，远程拷贝文件夹到其他节点，之后只需修改config/server.properties中三处，还有myid文件中的值。
4.创建myid文件，进入dataDir，创建myid文件，将三个节点上的myid文件分别写入1，2，3
—-myid是zk集群用来发现彼此的标识，必须创建，且不能相同；
5.启动zookeeper
方法一：
进入kafka目录 执行启动zookeeper命令：
./bin/zookeeper-server-start.sh config/zookeeper.properties &
三台机器都执行启动命令，查看zookeeper的日志文件，没有报错就说明zookeeper集群启动成功了。
方法二：
进入kafka目录 执行启动zookeeper命令：
./bin/zookeeper-server-start.sh config/zookeeper.properties
此时，窗口会卡住，没有回到命令提示符状态，如图：
不要退出，重新打开一个终端窗口，输入:jps
表示zookeeper已经在运行中。
另外两个节点同理。
6.启动kafka集群，
方法一：
进入kafka目录，执行如下命令 ：
./bin/kafka-server-start.sh –daemon config/server.properties &
三个节点均要启动；启动无报错，即搭建成功，可以生产和消费消息，来检测是否搭建成功。
方法二：
进入kafka目录 执行如下命令：
./bin/kafka-server-start.sh config/server.properties
此时，窗口会卡住，没有回到命令提示符状态，如图：
不要退出，重新打开一个终端窗口，输入:jps
表示Kafka已经在运行中。
另外两个节点同理。
7.检测是否搭建成功
可以生产和消费消息，来检测是否搭建成功。
7.1创建topic–test
bin/kafka-topics.sh --bootstrap-server 10.103.105.91:9092,10.103.105.92:9092,10.103.105.93:9092 --create --topic test1 --partitions 2 --replication-factor 1
如图所示，创建成功。
参数说明：
topic:指定topic name
partitions:指定分区数，这个参数需要根据broker数和数据量决定，正常情况下，每个broker上两个partition最好
replication-factor:副本数，建议设置为2
7.2列出已创建的topic列表
bin/kafka-topics.sh --list --bootstrap-server 10.103.105.91:9092
7.3模拟客户端去发送消息
./bin/kafka-console-producer.sh --broker-list 10.103.105.91:9092, 10.103.105.92:9092, 10.103.105.93:9092 --topic test1
如图，可在>之后输入信息，然后再次开启新的终端或者直接按CTRL+C退出。
7.4模拟客户端去接受消息
bin/kafka-console-consumer.sh --bootstrap-server 10.103.105.91:9092, 10.103.105.92:9092, 10.103.105.93:9092 --from-beginning --topic test1
此时，我们到其他节点上看是否可以接受到消息。
可以接受到，成功。
7.5查看指定的主题
./bin/kafka-topics.sh --describe --bootstrap-server 10.103.105.91:9092, 10.103.105.92:9092, 10.103.105.93:9092 --topic test1
8.关闭
./bin/kafka-server-stop.sh
./bin/zookeeper-server-stop.sh
参考文章：https://blog.csdn.net/zxy987872674/article/details/72493128
2-问题及解决办法：
2.1
解决办法：
在较新版本（2.2 及更高版本）的 Kafka 不再需要 ZooKeeper 连接字符串，即- -zookeeper localhost:2181。使用 Kafka Broker的 --bootstrap-server localhost:9092来替代- -zookeeper localhost:2181。
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic heima --partitions 2 --replication-factor 1
bin/kafka-topics.sh --bootstrap-server 10.103.105.91:9092 --create --topic heima --partitions 2 --replication-factor 1
2.2-创建topic失败
解决办法：
修改Kafka的server.properties
2.3 zookeeper正常启动，但Kafka启动不了
解决办法：删除/config/server.properties中所设置的log.dirs存储位置下所有文件。
2.4
解决办法：
bin/kafka-topics.sh --bootstrap-server 10.103.105.91:9092 --create --topic heima --partitions 2 --replication-factor 1
2.5.创建topics时失败
原命令： bin/kafka-topics.sh --list --zookeeper localhost:2181 失败
解决办法：采用命令：
bin/kafka-topics.sh --list --bootstrap-server 10.103.105.91:9092