kafka-消费者-指定offset消费（SpringBoot整合Kafka）

文章目录

  • 1、指定offset消费
    • 1.1、创建消费者监听器‘
    • 1.2、application.yml配置
    • 1.3、使用 Java代码 创建 主题 my_topic1 并建立3个分区并给每个分区建立3个副本
    • 1.4、创建生产者发送消息
      • 1.4.1、分区0中的数据
    • 1.5、创建SpringBoot启动类
    • 1.6、屏蔽 kafka debug 日志 logback.xml
    • 1.7、引入spring-kafka依赖
    • 1.8、消费者控制台：

1、指定offset消费

1.1、创建消费者监听器‘

package com.atguigu.spring.kafka.consumer.listener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;
@Component
public class MyKafkaPartitionListener {//初始化偏移量指定后，每次重启都会从该位置消费一轮，所以一般是调式解决问题时才使用@KafkaListener(topicPartitions = {@TopicPartition(topic = "my_topic1",partitionOffsets = {@PartitionOffset(partition = "0",initialOffset = "2")})}, groupId = "my_group1")public void onMessage1(ConsumerRecord<String, String> record) {System.out.println("my_group1消费者1获取到消息：topic = "+ record.topic()+"，partition:"+record.partition()+"，offset = "+record.offset()+"，key = "+record.key()+"，value = "+record.value());}}

1.2、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 # 消费者提交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

1.3、使用 Java代码 创建 主题 my_topic1 并建立3个分区并给每个分区建立3个副本

package com.atguigu.spring.kafka.consumer.config;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
@Configuration
public class MyKafkaConfig {@Beanpublic NewTopic springTestPartitionTopic() {return TopicBuilder.name("my_topic1") //主题名称.partitions(3) //分区数量.replicas(3) //副本数量.build();}
}

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

1.4、创建生产者发送消息

package com.atguigu.spring.kafka.consumer;import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;@SpringBootTest
class SpringKafkaConsumerApplicationTests {@ResourceKafkaTemplate kafkaTemplate;@Testvoid contextLoads() {for (int i = 0; i < 10; i++) {kafkaTemplate.send("my_topic1",i%3,"", "指定分区消费"+i);}}}

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

1.4.1、分区0中的数据

[[{"partition": 0,"offset": 0,"msg": "指定offset消费0","timespan": 1717660785962,"date": "2024-06-06 07:59:45"},{"partition": 0,"offset": 1,"msg": "指定offset消费3","timespan": 1717660785974,"date": "2024-06-06 07:59:45"},{"partition": 0,"offset": 2,"msg": "指定offset消费6","timespan": 1717660785975,"date": "2024-06-06 07:59:45"},{"partition": 0,"offset": 3,"msg": "指定offset消费9","timespan": 1717660785975,"date": "2024-06-06 07:59:45"}]
]

1.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);}}

1.6、屏蔽 kafka debug 日志 logback.xml

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

1.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>

1.8、消费者控制台：

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

此时如果重新启动 SpringKafkaConsumerApplication 消费者还是会消费数据，重复消费

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

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

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

相关文章

服务器数据恢复—EqualLogic存储硬盘灯亮黄色的数据恢复案例

服务器数据恢复—EqualLogic存储硬盘灯亮黄色的数据恢复案例

服务器数据恢复环境&#xff1a; 一台某品牌EqualLogic PS 6011型号存储&#xff0c;底层有一组由16块SAS硬盘组建的RAID5阵列&#xff0c;上层存储空间划分了4个卷&#xff0c;格式化为VMFS文件系统&#xff0c;存放虚拟机文件。 服务器故障&#xff1a; 存储设备上两块硬盘指…
阅读更多...
基于STC89C52单片机空气PM2.5系统设计资料

基于STC89C52单片机空气PM2.5系统设计资料

#include <reg52.h>#include <intrins.h>#define uint unsigned int#define uchar unsigned char //宏定义sbit RSP1^6;//液晶接口sbit ENP1^7;sbit LED P2^0;//粉尘传感器控制接口sbit ADCS P3^7;//AD0832接口sbit ADCLK P3^5;sbit ADDI P3^6;sbit ADDO P3^6;…
阅读更多...
【大模型】基于Hugging Face调用及微调大模型（1）

【大模型】基于Hugging Face调用及微调大模型（1）

文章目录 一、前言二、Transformer三、Hugging Face3.1 Hugging Face Dataset3. 2 Hugging Face Tokenizer3.3 Hugging Face Transformer3.4 Hugging Face Accelerate 四、基于Hugging Face调用模型4.1 调用示例4.2 调用流程概述4.2.1 Tokenizer4.2.2 模型的加载4.2.3 模型基本…
阅读更多...
# RocketMQ 实战：模拟电商网站场景综合案例（二）

# RocketMQ 实战：模拟电商网站场景综合案例（二）

RocketMQ 实战&#xff1a;模拟电商网站场景综合案例&#xff08;二&#xff09; 一、SpringBoot 整合 Dubbo &#xff1a;dubbo 概述 1、dubbo 概述 Dubbo &#xff1a;是阿里巴巴公司开源的一款高性能、轻量级的 Java RPC 框架&#xff0c;它提供了三大核心能力&#xff1a…
阅读更多...
【WP】猿人学15_备周则意怠_常见则不疑

【WP】猿人学15_备周则意怠_常见则不疑

https://match.yuanrenxue.cn/match/15 抓包分析 抓包分析有一个m参数&#xff0c;三个数字组成 追栈/扣代码 根据启动器顺序追栈&#xff0c;一般优先跳过 jQuery 直接能找到加密函数 每次获取的数字都不一样 window.m function() { t1 parseInt(Date.parse(new Date(…
阅读更多...
【全开源】Java共享茶室棋牌室无人系统支持微信小程序+微信公众号

【全开源】Java共享茶室棋牌室无人系统支持微信小程序+微信公众号

打造智能化休闲新体验 一、引言&#xff1a;智能化休闲时代的来临 随着科技的飞速发展&#xff0c;智能化、无人化服务逐渐渗透到我们生活的各个领域。在休闲娱乐行业&#xff0c;共享茶室棋牌室无人系统源码的出现&#xff0c;不仅革新了传统的休闲方式&#xff0c;更为消费…
阅读更多...
Marin说PCB之如何在主板上补偿链路中的走线的等长误差？

Marin说PCB之如何在主板上补偿链路中的走线的等长误差？

一场雨把我困在这里&#xff0c;你冷漠地看我没有穿雨衣淋成落汤鸡。今天刚刚出门时候看天气预报没有雨&#xff0c;于是我就没有带雨衣骑电动车去公司了&#xff0c;谁知道回来的路上被淋成狗了。天气预报就像是女人的脾气那样&#xff0c;不能完全相信的。 好了&#xff0c;我…
阅读更多...
什么是视频号招商团长？如何加入成为视频号招商团长

什么是视频号招商团长？如何加入成为视频号招商团长

视频号招商团长&#xff0c;是通过微信视频号平台的线上和线下活动&#xff0c;撮合商家和达人进行合作&#xff0c;帮助商家、达人在视频号成长发展&#xff1b;同时还可以通过邀请内容创作者入驻微信视频号并为其提供支持&#xff1b;从而获取佣金收益的&#xff0c;而其作用…
阅读更多...
【LeetCode算法】第100题：相同的树

【LeetCode算法】第100题：相同的树

目录 一、题目描述 二、初次解答 三、官方解法 四、总结 一、题目描述 二、初次解答 1. 思路&#xff1a;二叉树的先序遍历。采用递归的先序遍历方法&#xff0c;首先访问根节点若不同则返回false&#xff0c;其次访问左子树和右子树。在访问左右子树时&#xff0c;需要注意…
阅读更多...
CAN总线学习笔记-CAN帧结构

CAN总线学习笔记-CAN帧结构

数据帧 数据帧&#xff1a;发送设备主动发送数据&#xff08;广播式&#xff09; 标准格式的11ID不够用了&#xff0c;由此产生了扩展格式 SOF&#xff1a;帧起始&#xff0c;表示后面一段波形为传输的数据位 ID&#xff1a;标识符&#xff0c;区分功能&#xff0c;同时决定优…
阅读更多...
【qt】项目移植

【qt】项目移植

项目移植 一.前言二.同名问题三.具体操作1.修改文件名2.修改类名3.修改一些不能自动改的名4.修改.ui文件5.删除原来自动生成的ui_xxx.h文件6.修改头文件 四.导入项目五.使用导入的项目六.项目建议 一.前言 终于概率论考完了,有时间了,接着上个项目,我们继续来完成我们的多窗口开…
阅读更多...
探索 LLM 预训练的挑战，GPU 集群架构实战

探索 LLM 预训练的挑战，GPU 集群架构实战

万卡 GPU 集群实战&#xff1a;探索 LLM 预训练的挑战 一、背景 在过往的文章中&#xff0c;我们详细阐述了LLM预训练的数据集、清洗流程、索引格式&#xff0c;以及微调、推理和RAG技术&#xff0c;并介绍了GPU及万卡集群的构建。然而&#xff0c;LLM预训练的具体细节尚待进一…
阅读更多...
Qt——升级系列（Level Two）：Hello Qt 程序实现、项目文件解析、Qt 编程注意事项

Qt——升级系列（Level Two）：Hello Qt 程序实现、项目文件解析、Qt 编程注意事项

Hello Qt 程序实现 使用“按钮”实现 纯代码方式实现&#xff1a; // Widget构造函数的实现 Widget::Widget(QWidget *parent): QWidget(parent) // 使用父类构造函数初始化QWidget&#xff0c;传入父窗口指针, ui(new Ui::Widget) // 创建Ui::Widget类的实例&#xff0c;并…
阅读更多...
YOLOv8_obb预测流程-原理解析[旋转目标检测理论篇]

YOLOv8_obb预测流程-原理解析[旋转目标检测理论篇]

YOLOv8_obb的预测流程，主要分预处理模块、推理模块和后处理模块。这里面有很多内容是和目标检测预测流程是重合的，主要区别在于Angle分支、NMS后处理以及regularize_rboxes部分。本文也主要介绍一下这三个模块，其他模块可以结合YOLOv8预测流程-原理解析[目标检测理论篇]一起…
阅读更多...
Ffmpeg安装和简单使用

Ffmpeg安装和简单使用

Ffmpeg安装 下载并解压 进入官网 (https://ffmpeg.org/download.html)&#xff0c;选择 Window 然后再打开的页面中下滑找到 release builds&#xff0c;点击 zip 文件下载 环境变量配置 下载好之后解压&#xff0c;找到 bin 文件夹&#xff0c;里面有3个 .exe 文件 然后复制…
阅读更多...
Zookeeper复习

Zookeeper复习

一、入门 1、概念 zookeeper文件系统通知机制 2.特点 1&#xff09;、一个领导者&#xff0c;多个跟随者组成的集群。 2&#xff09;、集群中只要有半数以上存活机制&#xff0c;zookeeper集群能正产服务。zk适合安装奇数台。 3&#xff09;、全局数据一致&#xff1a;每…
阅读更多...
量化投资分析平台 迅投 QMT（四）获取标的期权的代码

量化投资分析平台 迅投 QMT（四）获取标的期权的代码

量化投资分析平台 迅投 QMT [迅投 QMT](https://www.xuntou.net/?user_code7NYs7O)我目前在使用有了底层标的如何获取期权的交易代码呢&#xff1f;上代码历史帖子 迅投 QMT 我目前在使用 两个月前&#xff08;2024年4月&#xff09;迅投和CQF有一个互动的活动&#xff0c;进…
阅读更多...
Windows defender 开启时无法访问共享文件夹，禁用时却可以的解决方法

Windows defender 开启时无法访问共享文件夹，禁用时却可以的解决方法

阅读更多...
5G+北斗智能手持终端在哪些行业中发挥作用

5G+北斗智能手持终端在哪些行业中发挥作用

在当今科技融合发展的浪潮中&#xff0c;5G北斗智能手持终端正逐步成为驱动各行各业智能化升级的关键力量。这一融合创新技术不仅重塑了传统的通信与定位方式&#xff0c;而且在多个核心领域展现了其变革性的应用价值。 5G北斗智能手持终端因其独特的技术组合&#xff0c;在多个…
阅读更多...
人工智能芯片封装技术及应用趋势分析

人工智能芯片封装技术及应用趋势分析

简介人工智能&#xff08;AI&#xff09;、物联网&#xff08;IoT&#xff09;和大数据的融合正在开创全新的智能时代&#xff0c;以智能解决方案改变各行各业。人工智能芯片在支持人工智能学习和推理计算方面发挥着非常重要的作用&#xff0c;可实现各行各业的多样化应用。 本…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023