【Kafka】2.在SpringBoot中使用官方原生java版Kafka客户端

在开始之前，需要先做点准备工作，用 IDEA 新建一个 Maven 项目，取名 kafka-study，然后删掉它的 src 目录，接着在 pom.xml 里面引入下面的依赖。这个项目的作用是作为整个工程的父项目，后面的项目都在此基础上新建 module 即可。

这里用到的环境信息如下：

• JDK1.8
• SpringBoot2.7.1
• IDEA
• Maven3.6.3

<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <packaging>pom</packaging>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.1</version>
        <relativePath/>
    </parent>

    <groupId>org.yuhuofei</groupId>
    <artifactId>kafka-study</artifactId>
    <version>1.0-SNAPSHOT</version>

    <name>kafka-study</name>
    <description>Study project for Kafka</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>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

1. 新建一个消息生产者

使用 IDEA 在 kafka-study 的基础上新建一个 Maven 类型的 module ，命名为 producer-01，直至完成。接着，我们改造这个 module ，把它变成我们想要的消息生产者。

1、在 java 目录下，新建一个包 com.yuhuofei ，然后再创建一个启动类 ProducerApplication.java ，内容如下:

package com.yuhuofei;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class ProducerApplication {

    public static void main(String[] args) {
        SpringApplication.run(ProducerApplication.class, args);
    }

}

2、在 resources 目录下，新建配置文件 application.properties ，内容如下：

server.port=8080
server.servlet.context-path=/producer-01
#Kafka配置
topic.name=my-kafka-topic

3、在自己的 pom.xml 文件中，引入 Kafka 客户端的依赖。这里需要注意的是，由于我们在上一篇博客里安装的是 kafka_2.11-2.2.1 版本的服务端，所以引入的 Kafka 客户端的版本最好对应，也选择 2.2.1 版本的。

<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>kafka-study</artifactId>
        <groupId>org.yuhuofei</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>producer-01</artifactId>

    <dependencies>
        <!--官方原生的java版kafka客户端依赖-->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.2.1</version>
        </dependency>
        <!--fastjson依赖-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>2.0.11.graal</version>
        </dependency>
    </dependencies>

</project>

4、由于 Kafka 进行消息的发布或者消费是需要有 Topic 的，因此我们需要要创建一个 Topic。在 zookeeper 和 Kafka 服务端都启动的情况下，打开 D:\Kafka\kafka_2.11-2.2.1\bin\windows 文件夹，在地址栏输入 cmd，打开控制台，然后用下面的命令创建、查看、删除 topic 。

如下图所示，我已经创建了一个名字为 my-kafka-topic 的 topic ，这个 topic 的分区为 1，副本也是 1。

# 创建topic
kafka-topics.bat --create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 1 --topic my-kafka-topic
# 查询topic列表
kafka-topics.bat --zookeeper 127.0.0.1:2181 --list					
# 删除topic
kafka-topics.bat --delete --zookeeper 127.0.0.1:2181 --topic my-kafka-topic
# 查询某个topic信息
kafka-topics.bat --describe --zookeeper localhost:2181 --topic my-kafka-topic
#kafka查看topic数据内容
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic my-kafka-topic --from-beginning

5、实现生产者发送消息

• controller层

package com.yuhuofei.controller;

import com.yuhuofei.service.ProducerInterface;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @Description
 * @ClassName ProducerController
 * @Author yuhuofei
 * @Date 2023/8/13 16:38
 * @Version 1.0
 */
@RequestMapping("/info")
@RestController
public class ProducerController {

    @Resource
    private ProducerInterface producerInterface;

    @GetMapping("/create-message")
    public void createTopic() {
        producerInterface.createMessage();
    }
}

• service层

package com.yuhuofei.service;

/**
 * @Description
 * @InterfaceName ProducerController
 * @Author yuhuofei
 * @Date 2023/8/13 16:39
 * @Version 1.0
 */
public interface ProducerInterface {

    void createMessage();
}

• serviceImpl层

package com.yuhuofei.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.yuhuofei.service.ProducerInterface;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @Description
 * @ClassName ProducerInterfaceImpl
 * @Author yuhuofei
 * @Date 2023/8/13 16:39
 * @Version 1.0
 */
@Service
public class ProducerInterfaceImpl implements ProducerInterface {

    @Value("${topic.name}")
    private String topicName;

    @Resource
    private Producer producer;

    @Override
    public void createMessage() {
        //创建测试数据
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("word", "这是生产者生产的测试消息");
        String jsonString = JSON.toJSONString(jsonObject);
        //使用kafka发送消息
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topicName, jsonString);
        producer.send(producerRecord);
    }
}

• 客户端配置类

package com.yuhuofei.config;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

/**
 * @Description 配置生产者客户端参数
 * @ClassName KafkaProperties
 * @Author yuhuofei
 * @Date 2023/8/13 22:18
 * @Version 1.0
 */
@Configuration
public class KafkaConfig {

    @Bean("producer")
    public Producer<String, String> getKafkaProducer() {
        Properties properties = new Properties();
        //配置Kafka的服务端IP和端口
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        //指定key使用的序列化类
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //指定value使用的序列化类
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        return new KafkaProducer<>(properties);
    }
}

6、启动服务，调用接口 http://localhost:8080/producer-01/info/create-message ，然后在控制台用命令查看一下消息是否写入，如下图所见，虽然是乱码，但是已经可以看出消息成功写入到 Kafka

到这里，消息生产者的创建暂时告一段落。

2. 新建一个消息消费者

和前面建立生产者类似的步骤，只是文件信息和名字不一样，这里步骤就不重复了，直接给出相关的配置和类。这个消费者，取名 consumer-01 。

• 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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>kafka-study</artifactId>
        <groupId>org.yuhuofei</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>consumer-01</artifactId>

    <dependencies>
        <!--官方原生的java版kafka客户端依赖-->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.2.1</version>
        </dependency>
        <!--fastjson依赖-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>2.0.11.graal</version>
        </dependency>
    </dependencies>
</project>

• properties文件

server.port=8081
server.servlet.context-path=/consumer-01
#Kafka配置
topic.name=my-kafka-topic

• 启动类 ConsumerApplication.java

package com.yuhuofei;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class ConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(ConsumerApplication.class, args);
    }

}

• 控制层ConsumerController.java

package com.yuhuofei.controller;

import com.yuhuofei.service.ConsumerInterface;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @Description
 * @ClassName ConsumerController
 * @Author yuhuofei
 * @Date 2023/8/14 0:19
 * @Version 1.0
 */
@RestController
@RequestMapping("/consumer")
public class ConsumerController {

    @Resource
    private ConsumerInterface consumerInterface;

    @GetMapping("/getData")
    public void getData() {
        consumerInterface.getData();
    }
}

• service层

package com.yuhuofei.service;

/**
 * @Description
 * @ClassName ConsumerInterface
 * @Author yuhuofei
 * @Date 2023/8/14 0:20
 * @Version 1.0
 */
public interface ConsumerInterface {
    void getData();
}

• serviceImpl层

package com.yuhuofei.service.impl;

import com.yuhuofei.service.ConsumerInterface;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;

/**
 * @Description
 * @ClassName ConsumerInterfaceImpl
 * @Author yuhuofei
 * @Date 2023/8/14 0:20
 * @Version 1.0
 */
@Service
public class ConsumerInterfaceImpl implements ConsumerInterface {

    @Value("${topic.name}")
    private String topicName;

    @Resource
    private Consumer consumer;

    @Override
    public void getData() {
        Collection<String> topics = Collections.singletonList(topicName);
        consumer.subscribe(topics);
        List<String> result = new ArrayList<>();
        while (true) {
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord consumerRecord : consumerRecords) {
                //从ConsumerRecord中获取消费数据
                String res = (String) consumerRecord.value();
                System.out.println("从Kafka中消费的原始数据: " + res);
                result.add(res);
                System.out.println(result);
            }
        }
    }
}

• config配置类

package com.yuhuofei.config;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

/**
 * @Description 配置消费者参数
 * @ClassName KafkaConsumerConfig
 * @Author yuhuofei
 * @Date 2023/8/14 0:08
 * @Version 1.0
 */
@Configuration
public class KafkaConsumerConfig {

    @Bean("consumer")
    public Consumer<String, String> getKafkaConsumer() {
        Properties properties = new Properties();
        //配置Kafka的服务端IP和端口
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        //指定key使用的序列化类
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //指定value使用的序列化类
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //指定消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-01-group");
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,100);
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
        return new KafkaConsumer<>(properties);
    }
}

最终整个项目目录如下图所示。

3. 测 试

1、检查 zookeeper 和 Kafka 服务端是否是启动的，没启动就先启动。

2、接着，启动消费者服务，然后调用一下接口 http://localhost:8081/consumer-01/consumer/getData ，这么做的目的启动一个线程，监听 topic 。

3、最后，启动生产者服务，调用接口 http://localhost:8080/producer-01/info/create-message ，向 Kafka 服务器写数据。

结果如下所示：

控制台看到的所有消息记录

生产者发送一条消息

消费者监听到一条消息并接收

在 SpringBoot 中，使用官方原生 java 版 Kafka 客户端的入门介绍到这里结束，还有更多使用方式，等待挖掘。