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

http://localhost:9988/

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

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:============> [测试值]