springboot中kafka的基本使用

一、 前言

在现在基本人手一部手机的时代，信息是无时不在，如何让信息在各个系统之间高效且准确的传递是一个很重要的技术，为此有大佬研究了一个个消息中间件技术，例如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讲解的资料，真得很好我就是从这上面学的,

详解 SpringBoot 集成 Kafka 全面实战，超全面精讲教程

Kafka3.x教程（从入门到调优，深入全面）