kafka 集群搭建

# zookeeper安装
第一步：使用FileZilla或者xftp可以通过ftp将zookeeper压缩包上传到192.168.100.136机器，解压，进入到 conf 目录，看到 zoo_sample.cfg 文件，cp 复制生成 zoo.cfg 文件，如下：

![](/media/202309/2023-09-09_031631_4662280.9963824600779511.png)

第二步，在192.168.100.136机器上，修改zoo.cfg，如下：
```
dataDir=/data/zookeeper/data 
dataLogDir=/data/zookeeper/logs

server.1=192.168.100.136:8880:7770 
server.2=192.168.100.137:8880:7770 
server.3=192.168.100.138:8880:7770
```

![](/media/202309/2023-09-09_031649_4532780.4300668445118405.png)
第三步：在192.168.100.136机器上，新建 /data/zookeeper/data 和 /data/zookeeper/logs 目录，因为 zoo.cfg 配置文件中指定了这两个目录，所以需要新建出来

![](/media/202309/2023-09-09_031700_0837310.5608717538207244.png)
第四步：将 zoo.cfg 复制到另外两个节点192.168.10.137、192.168.100.138上，并在这两个机器上新建 /data/zookeeper/data 和 /data/zookeeper/logs 目录。

第五步：分别在三个节点的 dataDir （即/data/zookeeper/data） 目录下创建myid文件，内容为：

第一个节点 192.168.100.136 为 1，

第二个节点 192.168.100.137 为 2，

第三个节点 192.168.100.138 为 3。

注意：myid文件里的值是和刚才zoo.cfg文件配置的server.1 server.2 server.3 对应的，第一台服务器就是 192.168.100.136，第二台服务器就是 192.168.100.137，第三台服务器就是 192.168.100.138。

第六步：三台机器上，启动zookeeper，并查看zookeeper状态，确定zookeeper启动是否正常
```
启动zookeeper
./bin/zkServer.sh start
查看zookeeper状态
./bin/zkServer.sh status
```

注意，zk三个命令

停止： ./bin/zkServer.sh stop

启动： ./bin/zkServer.sh start

查看状态：./bin/zkServer.sh status

如果是启动第一个节点，查看状态如下，没关系，这是因为才启动一个节点，还没有形成集群的原因，用 ps -ef|grep zookeeper 可以看到进程启动了就行了。

![](/media/202309/2023-09-09_031818_4413780.9031750221879765.png)

三个节点zookeeper均启动之后，如下：

![](/media/202309/2023-09-09_031833_6936420.28274485866916776.png)
eg1: zookeeper的注册中心和配置中心只是zk分布式协调的两个功能而已，主要协调数据一致性；

eg2: zookeeper集群只能部署奇数个节点(2n+1)个节点，避免选主失败；

eg3: kafka生产消息不需要依赖zookeeper，只有消费消息的时候，需要zookeeper用来记录offset。

# kafka的安装

进入到 conf 目录，看到config/server.properties 文件，并做如下修改
```
broker.id=0 
listeners=PLAINTEXT://192.168.100.136:9092 
log.dirs=/data/kafka/kafka-logs 
zookeeper.connect=192.168.100.136:2181,192.168.100.137:2181,192.168.100.138:2181
```

![](/media/202309/2023-09-09_031916_3272720.6411619615386464.png)

同理，使用FileZilla或者xftp可以通过ftp将kafka压缩包上传到 192.168.100.137 机器，解压，进入到 conf 目录，看到config/server.properties 文件，并做如下修改
```
broker.id=1 
listeners=PLAINTEXT://192.168.100.137:9092 
log.dirs=/data/kafka/kafka-logs 
zookeeper.connect=192.168.100.136:2181,192.168.100.137:2181,192.168.100.138:2181
```

注意：这里的 broker.id 和 listeners 这两个需要修改，log.dirs 和 zookeeper.connect 不需要改变，三个节点 broker.id 不能重复，哪个机器配哪个broker.id，就和 zoo.cfg 文件保持一致即可，listener 这里配置自己的ip地址。

继续，使用FileZilla或者xftp可以通过ftp将kafka压缩包上传到 192.168.100.138 机器，解压，进入到 conf 目录，看到config/server.properties 文件，并做如下修改
```
broker.id=2 
listeners=PLAINTEXT://192.168.100.138:9092 
log.dirs=/data/kafka/kafka-logs 
zookeeper.connect=192.168.100.136:2181,192.168.100.137:2181,192.168.100.138:2181
```
也是 broker.id 和 listeners 需要改变，log.dirs 和 zookeeper.connect 不需要改变。

第二步：三个节点kafka启动

在 192.168.100.136、192.168.100.137、192.168.100.138 三个机器上，分别运行这句
```
./bin/kafka-server-start.sh config/server.properties &
```
额外注意：有的时候，zookeeper或kafka启动会失败，这个一般是zk日志中有老数据，删掉各个节点 /data/zookeeper/data 和 /data/zookeeper/logs 就好，一般只要 zookeeper 不出问题，kafka启动是不会出问题的。

第三步：创建topic（默认需要手动创建，可以修改配置文件让kafka自动创建不存在的topic）

创建一个名称为 test2 的 topic ，指定 test topic有一个分区，一个副本。
```
./bin/kafka-topics.sh --create --topic test2 --bootstrap-server 192.168.100.137:9092 --partitions 1 --replication-factor 1
```

注意理解一下分区数和副本数的意思：因为这个topic只设置存放一个分区，且每个分区一个副本，所以 1 * 1 = 1 ，分区名称就是 test2-0，如果是三个分区，每个分区三个副本，所以 3 *3 = 9 个，如图：

查看topic信息
```
./bin/kafka-topics.sh --describe --topic test2 --bootstrap-server 192.168.100.137:9092
```

![](/media/202309/2023-09-09_032120_5052300.5126504745218184.png)

![](/media/202309/2023-09-09_032125_9861720.5169452200206421.png)

额外注意： kafka 中 topic 的概念类似rabbitmq 中 exchage 的概念。不同的是，rabbitmq中，消息存在在queue队列中，生产者生产消息的时候，如果没有这个exchange，会自己创建一个并发送消息到rabbitmq，经过exchange中转到queue队列中；但是，kafka中，生产者生产消息的时候，如果没有这个topic，不会自己创建一个并完成发送消息到kafka的partition中，而是会直接报错，所以，需要先创建topic，再发送消息。

第四步：发送消息，用kafka自带的命名模拟消息发送
```
./bin/kafka-console-producer.sh --bootstrap-server 192.168.100.137:9092 --topic test2
```
第五步：消费消息，用kafka自带的命名模拟消息消费
```
./bin/kafka-console-consumer.sh --bootstrap-server 192.168.100.136:9092 --topic test2 --from-beginni
```

![](/media/202309/2023-09-09_032214_4562490.9860611413088611.png)

---
生产消息和消费消息都是需要指定topic的;
消费消息还需要指定从头开始消费还是从尾开始消费，指定参数–from-beginning从头开始.
---
eg: 消费位置也可以通过auto.offset.reset 参数来指定，如下：

(1) earliest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费。

(2) latest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据。

(3) none：topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常。

小结：kafka常见

./bin/kafka-server-start.sh 启动

./bin/kafka-server-stop.sh 停止

./bin/kafka-topics.sh topic创建与查看状态

./bin/kafka-console-producer.sh 生产

sh./bin/kafka-console-consumer.sh 消费