kafka-消费者服务搭建配置简单消费(SpringBoot整合Kafka)

server/2024/9/24 13:19:45/

文章目录

  • 1、使用efak 创建 主题 my_topic1 并建立6个分区并给每个分区建立3个副本
  • 2、创建生产者发送消息
  • 3、application.yml配置
  • 4、创建消费者监听器
  • 5、创建SpringBoot启动类
  • 6、屏蔽 kafka debug 日志 logback.xml
  • 7、引入spring-kafka依赖

1、使用efak 创建 主题 my_topic1 并建立6个分区并给每个分区建立3个副本

在这里插入图片描述

在这里插入图片描述
在这里插入图片描述在这里插入图片描述

2、创建生产者发送消息

[root@localhost ~]# kafka-console-producer.sh --bootstrap-server 192.168.74.148:9095,192.168.748:9096,192.168.74.148:9097 --topic my_topic1
>1
>2
>3
>

在这里插入图片描述

[[{"partition": 1,"offset": 0,"msg": "1","timespan": 1717592203289,"date": "2024-06-05 12:56:43"},{"partition": 1,"offset": 1,"msg": "2","timespan": 1717592204046,"date": "2024-06-05 12:56:44"},{"partition": 1,"offset": 2,"msg": "3","timespan": 1717592204473,"date": "2024-06-05 12:56:44"}]
]

3、application.yml配置

server:port: 8120# v1
spring:Kafka:bootstrap-servers: 192.168.74.148:9095,192.168.74.148:9096,192.168.74.148:9097consumer:# read-committed读事务已提交的消息 解决脏读问题isolation-level: read-committed # 消费者的事务隔离级别:read-uncommitted会导致脏读,可以读取生产者事务还未提交的消息# 消费者是否自动ack :true自动ack 消费者获取到消息后kafka提交消费者偏移量enable-auto-commit: true # ??????offset# 消费者提交ack时多长时间批量提交一次auto-commit-interval: 1000# 消费者第一次消费主题消息时从哪个位置开始auto-offset-reset: earliest  #指定Offset消费:earliest | latest | nonekey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

4、创建消费者监听器

package com.atguigu.spring.kafka.consumer.listener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class MyKafkaListener {@KafkaListener(topics ={"my_topic1"},groupId = "my_group1")public void onMessage(ConsumerRecord<String, String> record) {System.out.println("消费者获取到消息:topic = "+ record.topic()+",partition:"+record.partition()+",offset = "+record.offset()+",key = "+record.key()+",value = "+record.value());}}

5、创建SpringBoot启动类

package com.atguigu.spring.kafka.consumer;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;// Generated by https://start.springboot.io
// 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn
@SpringBootApplication
public class SpringKafkaConsumerApplication {public static void main(String[] args) {SpringApplication.run(SpringKafkaConsumerApplication.class, args);}}

kafka_debug__logbackxml_113">6、屏蔽 kafka debug 日志 logback.xml

<configuration>      <!-- 如果觉得idea控制台日志太多,src\main\resources目录下新建logback.xml
屏蔽kafka debug --><logger name="org.apache.kafka.clients" level="debug" />
</configuration>

kafka_123">7、引入spring-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 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>3.0.5</version><relativePath/> <!-- lookup parent from repository --></parent><!-- Generated by https://start.springboot.io --><!-- 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn --><groupId>com.atguigu</groupId><artifactId>spring-kafka-consumer</artifactId><version>0.0.1-SNAPSHOT</version><name>spring-kafka-consumer</name><description>spring-kafka-consumer</description><properties><java.version>17</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

此时启动SpringKafkaConsumerApplication,控制台会打印数据

  .   ____          _            __ _ _/\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \\\/  ___)| |_)| | | | | || (_| |  ) ) ) )'  |____| .__|_| |_|_| |_\__, | / / / /=========|_|==============|___/=/_/_/_/:: Spring Boot ::                (v3.0.5)消费者获取到消息:topic = my_topic1,partition:1,offset = 0,key = null,value = 1
消费者获取到消息:topic = my_topic1,partition:1,offset = 1,key = null,value = 2
消费者获取到消息:topic = my_topic1,partition:1,offset = 2,key = null,value = 3

如果此时重新启动SpringKafkaConsumerApplication,控制台将不会打印数据,因为已经消费过数据

  .   ____          _            __ _ _/\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \\\/  ___)| |_)| | | | | || (_| |  ) ) ) )'  |____| .__|_| |_|_| |_\__, | / / / /=========|_|==============|___/=/_/_/_/:: Spring Boot ::                (v3.0.5)

http://www.ppmy.cn/server/47974.html

相关文章

c语言项目-贪吃蛇项目2-游戏的设计与分析

文章目录 前言游戏的设计与分析地图&#xff1a;这里简述一下c语言的国际化特性相关的知识<locale.h> 本地化头文件类项setlocale函数 上面我们讲到需要打印★&#xff0c;●&#xff0c;□三个宽字符找到这三个字符打印的方式有两种&#xff1a; 控制台屏幕的长宽特性&a…

java web爬虫

目录 读取本地文件 从网站读取文件 java爬虫 总结 读取本地文件 import java.io.File; import java.io.PrintWriter; import java.util.Scanner;public class ReplaceText {public static void main() throws Ex

天润融通,荣获2024中国AI应用层创新企业

AI技术发展日新月异&#xff0c;可谓“AI一天&#xff0c;人间一年”。 从2023年到2024年&#xff0c;短短一年的时间&#xff0c;大模型技术的发展就已经逐步从追求“技术突破”转向了追求“应用落地”。如何将大模型的技术与企业的生产、运营、销售等场景结合起来&#xff0…

【保姆级讲解Outlook邮箱的使用技巧】

&#x1f3a5;博主&#xff1a;程序员不想YY啊 &#x1f4ab;CSDN优质创作者&#xff0c;CSDN实力新星&#xff0c;CSDN博客专家 &#x1f917;点赞&#x1f388;收藏⭐再看&#x1f4ab;养成习惯 ✨希望本文对您有所裨益&#xff0c;如有不足之处&#xff0c;欢迎在评论区提出…

flink 状态

状态&#xff08;State&#xff09;是一个重要的概念&#xff0c;它允许Flink在处理流数据时跟踪和存储中间结果。这对于实现复杂的计算逻辑和满足应用需求至关重要。 Working with State 1. 状态类型 Flink支持两种主要类型的状态&#xff1a; 1.1 算子状态&#xff08;Op…

好用软件推荐

软件功能相关介绍地址FastStone截图&#xff08;长截图、定时截图等&#xff09;CSDNhttps://www.faststone.org/FSCaptureDownload.htmQuicker快捷访问https://getquicker.net/https://getquicker.net/

深度学习 --- stanford cs231 编程作业(assignment1,Q2: SVM分类器)

stanford cs231 编程作业之SVM分类器 写在最前面&#xff1a; 深度学习&#xff0c;或者是广义上的任何学习&#xff0c;都是“行千里路”胜过“读万卷书”的学识。这两天光是学了斯坦福cs231n的一些基础理论&#xff0c;越往后学越觉得没什么。但听的云里雾里的地方也越来越多…

word 无法自动检测拼写

word 有时候不能分辨是哪种语言,比如把英语错认为法语 。 例如&#xff1a;Interlaayer spacace,发现误认为是法语。 1、选中Interlaayer spacace 2、点击语言下拉按钮 选择设置校对语言 发现校对语言为法语 3、手动修改校对语言为英语&#xff0c;并点击确认。 4、发现现…