Windows本地搭建Zookeeper和Kafka环境
Windows本地搭建Zookeeper和Kafka环境
Zookeeper
1.1 下载Zookeeper
下载地址
随便进一个站点，默认是新版本，旧版本点击archives进入，选择合适的版本下载，本文使用的是3.7.2
下载时候选择apache-zookeeper-3.7.2-bin.tar.gz 格式的，编译后的，解压后直接使用
1.2 ZK启动
1.2.1 单机启动
解压后进入 /zk/conf/ 复制一份zoo_sample.cfg文件到同级目录，然后更改名称为zoo.cfg。编辑zoo.cfg，确认端口和数据目录
clientPort=2181
dataDir=D:\\xxx\\apache-zookeeper-3.7.2-bin\\data
然后进入zk/bin/目录下，运行zkServer.cmd则启动zk服务器完成
1.2.2 伪集群启动
解压后拷贝3份，同理修改配置zoo.cfg，多了server.x的配置。区别在于端口要配置不一样，因为是一台电脑
dataDir指向自己的目录的data,每个data下面创建myid文件，myid就是下面配置的1,2,3 ,对应conf文件里面的server.x
举例：
D:\env\apache-zookeeper-3.7.2-bin-A\data\myid的内容就是 1
D:\env\apache-zookeeper-3.7.2-bin-B\data\myid的内容就是 2
D:\env\apache-zookeeper-3.7.2-bin-C\data\myid的内容就是 3
dataDir=D:\\env\\apache-zookeeper-3.7.2-bin-A\\data
clientPort=2181
server.1=localhost:2887:3887
server.2=localhost:2888:3888
server.3=localhost:2889:3889
dataDir=D:\\env\\apache-zookeeper-3.7.2-bin-B\\data
clientPort=2182
server.1=localhost:2887:3887
server.2=localhost:2888:3888
server.3=localhost:2889:3889
dataDir=D:\\env\\apache-zookeeper-3.7.2-bin-C\\data
clientPort=2183
server.1=localhost:2887:3887
server.2=localhost:2888:3888
server.3=localhost:2889:3889
启动同理，到3个zk的bin目录下启动zkserver，也可以写个bat一次性启动
zk-cluster.bat
@echo off
start cmd /c "D:\\env\\apache-zookeeper-3.7.2-bin-A\\bin\\zkServer.cmd"
start cmd /c "D:\\env\\apache-zookeeper-3.7.2-bin-B\\bin\\zkServer.cmd"
start cmd /c "D:\\env\\apache-zookeeper-3.7.2-bin-C\\bin\\zkServer.cmd"
Kafka
下载
下载地址
选择二进制文件下载，我下载的是2.13-3.71
下载后解压，进入config目录，修改server.properties文件
对接单zk ，这边要和上面zk的端口一样
zookeeper.connect=localhost:2181
log.dirs=D:\\env\\kafka-zk-single\\logs
对接zk集群，这边要和上面zkA、B、C的端口一样
zookeeper.connect=localhost:2181,localhost:2182,localhost:2183
log.dirs=D:\\env\\kafka-zk-cluster\\logs
启动Kafka
进入kakfa/bin/windows/目录，启动kafka-server-start.bat 同时指定刚才的配置文件
也可以写个bat，放在外面
kafka-single.bat
"D:\\env\\kafka\\bin\\windows\\kafka-server-start.bat" "D:\\env\\kafka\\config\\server.properties"
这里提醒一下，不要一个kafka一会对接zk单机，一会对接zk集群，集群id是唯一的，混着用会报错
可以再复制一个kafka出来
类似
kafka-zk-single 用来对接单zk
kafka-zk-cluster 用来对接zk集群
Kafka-UI
Kafka可视化监控，类似的软件很多，这里用UI for Apache Kafka
主要是方便，一个jar包，直接启动就好了，不过要求java版本高于17? 用idea下载一个，直接指定位置启动就好了，不影响日常jdk8使用
下载好后同目录创建一个配置文件 application.yml。
server:
port: 9988
kafka:
clusters:
-
name: local
bootstrapServers: localhost:9092
简单解释下，详细配置看文档
一个是修改下端口，默认是8080，可能会冲突
另一个就是配置kafka的地址
然后启动！
C:\Users\Admin\.jdks\graalvm-jdk-17.0.11\bin\java.exe -jar kafka-ui-api-v0.7.2.jar -Dspring.config.additional-location=application.yml
Spring Boot对接Kafka
加依赖、加配置、写代码
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.8.11</version>
</dependency>
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
auto-offset-reset: latest
enable-auto-commit: true
auto-commit-interval: 3000
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
producer:
retries: 0
acks: 1
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
3种监听方式
@KafkaListener(topics = "test", groupId = "test_group")
public void receiveMessage(String message) {
System.out.println("消费消息:============> " + message);
}
@KafkaListener(topics = "test", groupId = "test_group1", properties = {"auto.commit.interval.ms=4000"})
public void receiveMessage2(ConsumerRecord<String, String> record) {
System.out.println("消费record:============> " + record);
}
@KafkaListener(topics = "test", groupId = "test_group2")
public void receiveMessage3(List<ConsumerRecord<String, String>> records) {
System.out.println("消费records:============> " + records);
}
启动程序。。。
测试
进入web页面test的topic里面
代码控制台打印：
消费record:============> ConsumerRecord(topic = test, partition = 0, leaderEpoch = 0, offset = 1, CreateTime = 1724315931298, serialized key size = 9, serialized value size = 9, headers = RecordHeaders(headers = [], isReadOnly = false), key = 测试key, value = 测试值)
消费消息:============> 测试值
消费records:============> [测试值]