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
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
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"
下载
下载地址
选择二进制文件下载，我下载的是2.13-3.71
下载后解压，进入config目录，修改server.properties文件
对接单zk ，这边要和上面zk的端口一样
- zookeeper.connect=localhost:2181
- log.dirs=D:\\env\\kafka-zk-single\\logs
- zookeeper.connect=localhost:2181,localhost:2182,localhost:2183
- log.dirs=D:\\env\\kafka-zk-cluster\\logs
进入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出来
类似
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
springboot对接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
- @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:============> [测试值]
|
