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

<?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 {@Resourceprivate 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;@Resourceprivate Producer producer;@Overridepublic 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 {@Resourceprivate 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;@Resourceprivate Consumer consumer;@Overridepublic 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);}
}