Windows下快速使用kafka3.x版本与Spring的整合demo_window使用apache kafka3

一、安装下载kafka

目前最新版本是3.7.0，注意下，kafka自从3版本后就内置了zookeeper，后面启动的zookeeper服务是内置的zookeeper.

链接：https://pan.baidu.com/s/1BUR9Wr39YlrPK0diC8VDIQ?pwd=1234
提取码：1234

下载之后然后解压是这样的：

二、更改配置

1、新建一个logs文件夹

2、去config目录下去修改server.properties，修改的地方我用红框标注出来了

这里改为之前新建的logs目录
改完这些地方后，保存退出。

三、启动服务

1、先启动kafka内置的zookeeper服务，启动命令如下

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties
1

2、在启动kafka服务，命令如下

.\bin\windows\kafka-server-start.bat .\config\server.properties
1

注意，这两个命令要在bin当前目录下的控制台执行，先执行zookeeper命令，最小化窗口，然后在输入cmd，再打开一个窗口执行kafka命令，共两个窗口，如图所示。
一共两个窗口
zookeeper窗口：

kafka窗口：

四、整合Spring

先看一下目录结构，非常简单，照着这个搭建就行，以下给出具体代码参考。
1、pom.xml
全部放在这里了，需要什么依赖、插件，自取。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.example</groupId>
    <artifactId>kafka-demo-win</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>kafka-demo-win</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <spring-boot.version>2.6.13</spring-boot.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus</artifactId>
            <version>3.4.2</version>
        </dependency>
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-annotation</artifactId>
            <version>3.4.2</version>
        </dependency>
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>3.4.2</version>
        </dependency>
    </dependencies>
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>${spring-boot.version}</version>
                <configuration>
                    <mainClass>com.cfl.kafka.KafkaDemoWinApplication</mainClass>
                    <skip>true</skip>
                </configuration>
                <executions>
                    <execution>
                        <id>repackage</id>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

2、生成者EventProducer

package com.cfl.kafka.producer;




import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class EventProducer {

    //加入了spring-kafka依赖 + .yml配置信息，springboot自动配置好了kafka，自动装配好了KafkaTemplate这个Bean
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;


    public void sendEvent() {
        kafkaTemplate.send("hello-topic", "hello kafka");
    }


}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

3、消费者EventConsumer

package com.cfl.kafka.consumer;


import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class EventConsumer {
    @Autowired
    private ObjectMapper objectMapper;

    //采用监听的方式接收事件（消息、数据）
    @KafkaListener(topics = { "hello-topic"}, groupId = "hello-group")
    public void onEvent(String event) {
        System.out.println("读取到的事件：" + event);
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

4、application.yml配置文件

spring:
  application:
    #应用名称
    name: kafka-demo
  datasource:
    driver-class-name: com.mysql.jdbc.Driver
    url: jdbc:mysql://localhost:3306/test
    username: root
    password: password
  #kafka连接地址（ip+port）
  kafka:
    bootstrap-servers: 127.0.0.1:9092
  

    #配置生产者 (有24个配置)
#    producer:
#      #value默认是StringSerializer.class序列化，
#      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
#      #key默认是StringSerializer.class序列化，
#      key-serializer: org.apache.kafka.common.serialization.StringSerializer
#
#    #配置消费者(有24个配置)
#    consumer:
#      auto-offset-reset: latest
#
#    #配置模板默认的主题topic名称
#    template:
#      default-topic: default-topic
server:
  port: 18484

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

5、测试类KafkaDemoWinApplicationTests

package com.cfl.kafka;

import com.cfl.kafka.producer.EventProducer;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class KafkaDemoWinApplicationTests {
    @Autowired
    private EventProducer eventProducer;
    @Test
    void testEventProducer() {
        eventProducer.sendEvent();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

六、运行测试

给出以上的目录结构、代码及其配置基本差不多了。接下来就是运行测试。
1、先启用KafkaDemoWinApplication
2、运行测试类

运行测试类成功后，可以在application控制台看到输出：读取到的事件:hello kafka，表明整合成功。

七、总结与分享

整体来看，都还算是比较简单的，使用kafka呢，主要不是我想用，是我公司想搞，然后就让我研究研究，所以写一个demo出来，分享一下，时间长了，万一忘了，还可以回来看看。另一方面呢，就是希望大家也多多分享，个人在编程路上也是遇到很多问题，都是各种免费文章帮我解决问题，渡过各种艰难险阻，所以很感谢许多免费创作者。这也是我写文章的初衷，算是一种传承和接力吧。