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

目 录

    • 1. 新建一个消息生产者
    • 2. 新建一个消息消费者
    • 3. 测 试

在开始之前，需要先做点准备工作，用 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 {@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);}
}

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

在这里插入图片描述

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 客户端的入门介绍到这里结束，还有更多使用方式，等待挖掘。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/70598.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

[oneAPI] 手写数字识别-LSTM

[oneAPI] 手写数字识别-LSTM

[oneAPI] 手写数字识别-LSTM 手写数字识别参数与包加载数据模型训练过程结果 oneAPI 比赛&#xff1a;https://marketing.csdn.net/p/f3e44fbfe46c465f4d9d6c23e38e0517 Intel DevCloud for oneAPI&#xff1a;https://devcloud.intel.com/oneapi/get_started/aiAnalyticsToolk…
阅读更多...
在本地搭建WAMP服务器并通过端口实现局域网访问（无需公网IP）

在本地搭建WAMP服务器并通过端口实现局域网访问（无需公网IP）

文章目录 前言1.Wamp服务器搭建1.1 Wamp下载和安装1.2 Wamp网页测试 2. Cpolar内网穿透的安装和注册2.1 本地网页发布2.2 Cpolar云端设置2.3 Cpolar本地设置 3. 公网访问测试4. 结语 前言 软件技术的发展日新月异&#xff0c;各种能方便我们生活、工作和娱乐的新软件层出不穷&a…
阅读更多...
第57步 深度学习图像识别：CNN可视化（Pytorch）

第57步 深度学习图像识别：CNN可视化（Pytorch）

基于WIN10的64位系统演示 一、写在前面 由于不少模型使用的是Pytorch&#xff0c;因此这一期补上基于Pytorch实现CNN可视化的教程和代码&#xff0c;以SqueezeNet模型为例。 二、CNN可视化实战 继续使用胸片的数据集&#xff1a;肺结核病人和健康人的胸片的识别。其中&…
阅读更多...
考研算法第46天： 字符串转换整数 【字符串，模拟】

考研算法第46天： 字符串转换整数 【字符串，模拟】

题目前置知识 c中的string判空 string Count; Count.empty(); //正确 Count ! null; //错误c中最大最小宏 #include <limits.h>INT_MAX INT_MIN 字符串使用发运算将字符加到字符串末尾 string Count; string str "liuda"; Count str[i]; 题目概况 AC代码…
阅读更多...
【数据分析入门】Numpy进阶

【数据分析入门】Numpy进阶

目录 一、数据重塑1.1 透视1.2 透视表1.3 堆栈/反堆栈1.3 融合 二、迭代三、高级索引3.1 基础选择3.2 通过isin选择3.3 通过Where选择3.4 通过Query选择3.5 设置/取消索引3.6 重置索引3.6.1 前向填充3.6.2 后向填充 3.7 多重索引 四、重复数据五、数据分组5.1 聚合5.2 转换 六、…
阅读更多...
腾讯云GPU服务器GN7实例NVIDIA T4 GPU卡

腾讯云GPU服务器GN7实例NVIDIA T4 GPU卡

腾讯云GPU服务器GN7实例搭载1颗 NVIDIA T4 GPU&#xff0c;8核32G配置&#xff0c;系统盘为100G 高性能云硬盘&#xff0c;自带5M公网带宽&#xff0c;系统镜像可选Linux和Windows&#xff0c;地域可选广州/上海/北京/新加坡/南京/重庆/成都/首尔/中国香港/德国/东京/曼谷/硅谷…
阅读更多...
机器学习---对数几率回归

机器学习---对数几率回归

1. 逻辑回归 逻辑回归&#xff08;Logistic Regression&#xff09;的模型是一个非线性模型&#xff0c; sigmoid函数&#xff0c;又称逻辑回归函数。但是它本质上又是一个线性回归模型&#xff0c;因为除去sigmoid映射函 数关系&#xff0c;其他的步骤&#xff0c;算法都是…
阅读更多...
JVM——HotSpot的算法细节实现

JVM——HotSpot的算法细节实现

一、根节点枚举 固定可作为GC Roots的节点主要在全局性的引用&#xff08;如常量或类静态属性&#xff09;与执行上下文&#xff08;如栈帧中的本地变量表&#xff09;中&#xff0c;尽管目标明确&#xff0c;但查找要做到高效很难。现在java应用越来越庞大&#xff0c;光方法区…
阅读更多...
Mr. Cappuccino的第60杯咖啡——Spring之BeanFactory和ApplicationContext

Mr. Cappuccino的第60杯咖啡——Spring之BeanFactory和ApplicationContext

Spring之BeanFactory和ApplicationContext 类图BeanFactory概述功能项目结构项目代码运行结果总结 ApplicationContext概述功能MessageSource&#xff08;国际化的支持&#xff09;概述项目结构项目代码运行结果 ResourcePatternResolver&#xff08;匹配资源路径&#xff09;概…
阅读更多...
从源代码编译构建Hive3.1.3

从源代码编译构建Hive3.1.3

从源代码编译构建Hive3.1.3 编译说明编译Hive3.1.3更改Maven配置下载源码修改项目pom.xml修改hive源码修改说明修改standalone-metastore模块修改ql模块修改spark-client模块修改druid-handler模块修改llap-server模块修改llap-tez模块修改llap-common模块 编译打包异常集合异常…
阅读更多...
ViewUI表格Table嵌套From表单-动态校验数据合法性的解决方法

ViewUI表格Table嵌套From表单-动态校验数据合法性的解决方法

项目场景&#xff1a; 项目需求&#xff1a;在表格中实现动态加减数据&#xff0c;并且每行表格内的输入框&#xff0c;都要动态校验数据&#xff0c;校验不通过&#xff0c;不让提交数据&#xff0c;并且由于表格内部空间较小&#xff0c;我仅保留红边框提示&#xff0c;文字…
阅读更多...
【Java转Go】快速上手学习笔记（二）之基础篇一

【Java转Go】快速上手学习笔记（二）之基础篇一

目录 创建项目数据类型变量常量类型转换计数器键盘交互流程控制代码运算符 创建项目 上篇我们安装好了Go环境&#xff0c;和用IDEA安装Go插件来开发Go项目&#xff1a;【Java转Go】快速上手学习笔记&#xff08;一&#xff09;之环境安装篇 。 这篇我们开始正式学习Go语言。我…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023