一、 前言
在现在基本人手一部手机的时代，信息是无时不在，如何让信息在各个系统之间高效且准确的传递是一个很重要的技术，为此有大佬研究了一个个消息中间件技术，例如RabbitMQ、ActiveMQ、Kafka、RocketMQ等等，比较而言，RabbitMQ和Kafka的使用居多，在此来学习一下如何使用他，写下此博客记录便于回顾，写得不够好请大家多多支持。
二、前提准备
已经搭建好kafka环境，本人搭建使用的 zookeeper-3.5.5 + kafka_2.12-3.3.1
三、主体内容
1. 创建项目
首先使用idea工具创建出一个springboot项目，通过在idea中选择组件kafka即可自动引入相关依赖
kafka项目创建.png
此处我使用的springboot版本是 2.7.10 对应的kafka客户端版本可自行在maven依赖中查看。
项目创建成功后pom文件内容如下
<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>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.10</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.zx</groupId>
<artifactId>springboot-kafka</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>springboot-kafka</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!--kafka依赖-->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
相较于平常创建的web项目只是多了 个spring-kafka 包，这个是官方的spring兼容包和springboot是无缝对接了。
2. 第二步就是对kafka进行配置
kafka使用时是区分 producer（生产者）和consumer（消费者）的，在此项目中是可以同时进行两者的配置使用的。对kafka的配置可以分为三部分
- 第一部分：kafka的连接配置
# kafka配置
spring.kafka.bootstrap-servers=192.168.1.6:9092
- 第二部分：对producer的设置
# producer配置
## 消息key序列化器
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.IntegerSerializer
## 消息value 序列化器
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
## 生产者每个批次最多放多少条记录
spring.kafka.producer.batch-size=16384
## 生产者一端总的可用发送缓冲区大小 32MB=33554432
spring.kafka.producer.buffer-memory=33554432
- 第三部分：对consumer设置
# cosumer配置
## 消息key反序列化器
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
## 消息value 反序列化器
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.group-id=consumer01
## 如果在kafka中找不到当前消费者的偏移量，则直接将偏移量重置为最早的
spring.kafka.consumer.auto-offset-reset=earliest
## 消费者的偏移量是自动提交还是手动提交
spring.kafka.consumer.enable-auto-commit=true
## 消费者偏移量自动提交的时间间隔
spring.kafka.consumer.auto-commit-interval=1000
配置完成了，我们启动一下项目，检验一下是否能正常启动，如果自行配置的某些项存在问题，项目是不能正常启动的。
3. 编写推送消息代码小demo
代码根据场景可分为三部分，分别为同步推送消息、异步推送消息、消息消费
同步推送消息
我们创建一个http接口来模拟同步消息推送
/**
* 同步
* @return
*/
@RequestMapping("sync/send/{message}")
public String sendSync(@PathVariable String message){
LOGGER.info("同步发送推送消息到kafka开始，topic = {}",demoTopic);
ListenableFuture<SendResult<Integer, String>> future = template.send(demoTopic, 0, 0, message);
try {
SendResult<Integer, String> sendResult = future.get();
RecordMetadata metadata = sendResult.getRecordMetadata();
LOGGER.info("推送结果：metadata: topic={},分区（partition）={}，偏移量（offset）={}",metadata.topic(),metadata.partition(),metadata.offset());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
return "success";
}
异步推送消息
同样异步也是使用一个接口模拟
/**
* 异步
* @return
*/
@RequestMapping("async/send/{message}")
public String sendAsync(@PathVariable String message){
ListenableFuture<SendResult<Integer, String>> future = template.send(demoTopic, 0, 0, message);
future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
@Override
public void onFailure(Throwable ex) {
LOGGER.error("发送消息失败,topic={},data:{}",demoTopic,message);
}
@Override
public void onSuccess(SendResult<Integer, String> result) {
RecordMetadata metadata = result.getRecordMetadata();
LOGGER.info("发送消息成功：metadata: topic={},分区（partition）={}，偏移量（offset）={}",metadata.topic(),metadata.partition(),metadata.offset());
LOGGER.info("发送成功消息：{}",message);
}
});
return "success";
}
消息消费
现在推送的消息有了，那就需要使用消息了，是如何使用呢，是通过注解 @KafkaListener来实现的
package com.zx.springboot.component;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
/**
* @author : Mr.Zx
* @version : 1.0.0
* @ClassName : ConsumerMessage.java
* @Description : TODO
* @createTime : 2023-04-09 09:51
*/
@Component
public class ConsumerMessage {
private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerMessage.class);
@Value("${message.demoTopic}")
private String listerTopic;
@KafkaListener(topics = "demoTopic-01")
public void onMessage(ConsumerRecord<Integer,String> record){
LOGGER.info("消费者收到消息：topic = {},partition={}，offset={}，key={}，value={}",record.topic(),record.partition(),record.offset(),record.key(),record.value());
}
}
基本使用代码写完了，就得跑起来测试一下了，最后测试结果不成功那是无意义的，上述使用场景比较简单，对于发送消息和消费消息的其他模式后续再进行完善
四、扩展
推荐两个关于kafka讲解的资料，真得很好我就是从这上面学的,
