使用Spring Boot和Kafka实现消息订阅和发送

文章目录

一，新建Spring Boot
- 1，Maven配置
- 2，无法识别为SpringBoot项目
- 3，无效的源发行版
- 4，无法访问SpringApplication
- 5，运行直接Finish
- 6，服务运行成功
二，安装启动Kafka
- 1，下载
- 2，配置
- 3，启动
- 4，其他命令
三，订阅发送消息
- 1，加入依赖
- 2，yam配置文件
- 3，消费者
- 4，生产者
- 5，接口
- 6，测试结果

一，新建Spring Boot

最近忙着搞低代码开发，好久没新建spring项目了，结果今天心血来潮准备建个springboot项目
在这里插入图片描述
注意Type选Maven，java选8，其他默认

1，Maven配置

点下一步后完成就新建了一个spring boot项目，配置下Maven环境，主要是settings.xml文件，里面要包含阿里云仓库，不然可能依赖下载不下来
在这里插入图片描述

2，无法识别为SpringBoot项目

在maven配置没问题的前提下，IDEA无法识别这是一个Spring Boot项目，倒腾半天，终于发现问题原因所在=======>是Maven版本太高的原因
在这里插入图片描述
把.mvn/wrapper目录下的maven-wrapper.properties文件第一行的版本号降低，比如说降为3.5.4，然后重新点下Maven的同步按钮

3，无效的源发行版

接下来运行项目报错：java: 无效的源发行版: 14
在这里插入图片描述
修改pom.xml中java.version值为8，原来是17

	<properties><java.version>17</java.version></properties>

4，无法访问SpringApplication

继续运行，继续报错在这里插入图片描述
降低spring-boot-starter-parent版本，原来是3.1.3，改为2.7.2

5，运行直接Finish

继续运行，没报错，服务执行Finish
在这里插入图片描述
需要添加web依赖

 		<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency>

6，服务运行成功

终于，一个空的spring boot项目成功跑起来了，喜极而泣
在这里插入图片描述

二，安装启动Kafka

1，下载

官网=>https://kafka.apache.org/downloads，下载最新版的kafka，目前是3.5.1
在这里插入图片描述

2，配置

解压到D盘Config目录下即完成安装，目录为D:\Config\kafka_2.13-3.5.1
修改配置文件
(1) server.properties

broker.id=1
log.dirs=/Config/kafka_2.13-3.5.1/logs-kafka

(2) zookeeper.properties

dataDir=/Config/kafka_2.13-3.5.1/logs-zookeeper

3，启动

先启动zookeeper

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

再启动kafka

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

停止的时候，先停止kafka，再停止zookeeper，直接ctrl+c停止

4，其他命令

1，查看topic列表

bin\windows\kafka-topics.bat --list --bootstrap-server localhost:9092

2，查看topic具体信息

bin\windows\kafka-topics.bat --describe --bootstrap-server localhost:9092 --topic test

3，创建topic

bin\windows\kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

三，订阅发送消息

1，加入依赖

 		<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

2，yam配置文件

application.yaml

spring:profiles:active: dev

application-dev.yaml

server:port: 8082servlet:context-path: /test-kafkaspring:cache:type: ehcacheconfig: classpath:ehcache.xmljpa:database-platform: com.enigmabridge.hibernate.dialect.SQLiteDialectkafka:bootstrap-servers: 127.0.0.1:9092consumer:group-id: kafka-demo-kafka-groupkey-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer producer:key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer retries: 10

3，消费者

@Slf4j
@Component
public class KafkaConsumer {@KafkaListener(topics = {"test_topic"})public void handlerMsg(String content) {log.info("接收到消息：消息值：{} ",content);}
}

4，生产者

@Slf4j
@Component
public class KafkaProducer {@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;public String sendMessage(String content) {String topic = "test_topic";kafkaTemplate.send(topic, content);return "发送成功";}
}

5，接口

@Slf4j
@RestController
public class KafkaController {@Autowiredprivate KafkaProducer kafkaProducer;@PostMapping("/sendMessage")public String sendMessage(@RequestParam String content) {kafkaProducer.sendMessage(content);return "ok";}
}