当前位置: 首页 > news >正文

kafka:消费者从指定时间的偏移开始消费(二)

我的前一篇博客《kafka:AdminClient获取指定主题的所有消费者的消费偏移(一)》为了忽略忽略掉上线之前的所有消息,从获取指定主题的所有消费者的消费偏移并计算出最大偏移来解决此问题。
但这个方案需要使用不常用的AdminClient类,而且如果该主题如果是第一次被消费者拉取消息时,因为得不到消费者的消费偏移,最后的结果,就是从0偏移开始拉取所有消息。并不能真正实现忽略上线之前所有消息的目的。
所以我又优化了方案。基本的原理就是使用KafkaConsumer.offsetsForTimes方法获取消费者的所有主题分区的指定时间的偏移,并将这个偏移作为消费开始的偏移(KafkaConsumer.seek方法) 。

	@Testpublic void test3SeekToTime() {// 配置Kafka消费者的属性Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("group.id", "my_consumer_group");props.put("key.deserializer", StringDeserializer.class.getName());props.put("value.deserializer", StringDeserializer.class.getName());// 创建Kafka消费者实例try(Consumer<String, String> consumer = new KafkaConsumer<>(props)){			boolean seek = false;/** * 循环开始的时间,* 忽略该时间之前的消息*/long startMills = System.currentTimeMillis();while (true) {try {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(4000));if(!seek) {if(!records.isEmpty()) {/** * 获取第一批消息时更新消息偏移到循环开始的时间*/consumer.offsetsForTimes(Maps.asMap(consumer.assignment(),t->startMills)).forEach((k,v)->{if(null != v) {System.out.println("seek %s to %s",k,v.offset());consumer.seek(k,v.offset());}});seek = true;}/** 跳过第一批获取到的消息,继续循环 */continue;}records.forEach(record -> {String value = record.value();System.out.println("Received message: " + value);});}catch (Exception e) {e.printStackTrace();}}}}

相关文章:

kafka:消费者从指定时间的偏移开始消费(二)

我的前一篇博客《kafka:AdminClient获取指定主题的所有消费者的消费偏移(一)》为了忽略忽略掉上线之前的所有消息&#xff0c;从获取指定主题的所有消费者的消费偏移并计算出最大偏移来解决此问题。 但这个方案需要使用不常用的AdminClient类&#xff0c;而且如果该主题如果是第…...

Spring的加载配置文件、容器和获取bean的方式

&#x1f40c;个人主页&#xff1a; &#x1f40c; 叶落闲庭 &#x1f4a8;我的专栏&#xff1a;&#x1f4a8; c语言 数据结构 javaweb 石可破也&#xff0c;而不可夺坚&#xff1b;丹可磨也&#xff0c;而不可夺赤。 Spring配置文件和容器相关 一、加载properties文件1.1加载…...

(二)利用Streamlit创建第一个app——单页面、多页面

1 单页面app Step1&#xff1a;创建一个新的Python脚本。我们称之为uber_pickups.py。 Step2&#xff1a;在您喜爱的IDE或文本编辑器中打开uber_pickups.py&#xff0c;然后添加以下行&#xff1a; import streamlit as st import pandas as pd import numpy as npStep3&…...

一条sql查询语句在mysql中的执行过程是什么

mysql的连接器 我们想要在mysql中执行一条sql查询语句&#xff0c;首先需要连接到mysql服务&#xff0c;那么客户端首先要向mysql服务端发起连接请求&#xff0c;我们可以在客户端用mysql -h [ip] -P [port] -u 用户名 -p 密码 命令向服务端发起连接请求&#xff0c;这个连接请…...

网络互联究竟是需要什么协议相同,什么协议不同?

混淆概念的几个说法&#xff1a; 说法1&#xff1a; “以太网交换机不可以实现采用不同网络层协议的互联” 原因&#xff1a;以太网交换机是数据链路层的设备&#xff0c;不懂网络层的知识 说法2&#xff1a; “网桥可互联不同的物理层、不同的MAC子层以及不同速率的以太网”…...

ajax axios json

目录 一、ajax概述 1. 概念 2. 实现方式 &#xff08;1&#xff09;原生的JS实现方式&#xff08;了解&#xff09; &#xff08;2&#xff09; JQeury实现方式 二、axios 介绍 三、axios使用 1. axios 发送get/post请求 2. axios验证用户名称是否存在 四、json 1. …...

外观模式——提供统一入口

1、简介 1.1、概述 在软件开发中&#xff0c;有时候为了完成一项较为复杂的功能&#xff0c;一个类需要和多个其他业务类交互&#xff0c;而这些需要交互的业务类经常会作为一个完整的整体出现&#xff0c;由于涉及的类比较多&#xff0c;导致使用时代码较为复杂。此时&#…...

Vue中导入并读取Excel数据

在工作中遇到需要前端上传excel文件获取到相应数据处理之后传给后端并且展示上传文件的数据. 一、引入依赖 npm install -S file-saver xlsxnpm install -D script-loadernpm install xlsx二、在main.js中引入 import XLSX from xlsx三、创建vue文件 <div><el-uplo…...

CUDA常用函数

cudaDeviceSynchronize cudaDeviceSynchronize是一个CUDA函数&#xff0c;用于同步当前设备上的所有CUDA流。它会阻塞调用它的线程&#xff0c;直到所有设备上的CUDA流都执行完为止。这可以确保在进行后续的CUDA操作时&#xff0c;先前的操作已经完成。 在CUDA程序中&#xff0…...

72. ElasticSearch常用命令

索引管理 1新建索引 curl -XPUT http://10.42.172.35:9200/index012 读写权限 curl -XPUT -d {"blocks.read":false} http://10.42.172.35:9200/index01/_settings3 查看索引 单个 curl -XGET http://10.42.172.35:9200/index01/_settings多个 curl -XGET http…...

2023.7.26(同余方程的通解与特解)

Water(扩欧求特解与通解) 题意&#xff1a;给容量分别为A与B的水杯&#xff0c;问确切喝到C水的最小操作次数 有4种操作&#xff1a;选一杯全喝&#xff0c;选一杯全部倒掉&#xff0c;选一杯装满&#xff0c;将一杯的水尽量倒到另一杯中 思路&#xff1a;只有AxByC有解时才能确…...

Diffusion扩散模型学习3——Stable Diffusion结构解析-以图像生成图像(图生图,img2img)为例

Diffusion扩散模型学习3——Stable Diffusion结构解析-以图像生成图像&#xff08;图生图&#xff0c;img2img&#xff09;为例 学习前言源码下载地址网络构建一、什么是Stable Diffusion&#xff08;SD&#xff09;二、Stable Diffusion的组成三、img2img生成流程1、输入图片编…...

LangChain||什么是LangChain? LangChain有什么用?

从Auto-GPT说起&#xff1a; Auto-GPT可以调用本地电脑工具处理复杂信息;Auto-GPT可以围绕目标查阅资 料、“独立思考”、及时反馈、并 及时调整下一步操作…Auto-GPT的诞生&#xff0c;创造了大家 对“将LLM作为智慧大脑来高效 处理综合复杂任务”的想象;首次尝试串联大语言模…...

秋招算法备战第28天 | 93.复原IP地址、78.子集、90.子集II

93. 复原 IP 地址 - 力扣&#xff08;LeetCode&#xff09; 这个问题可以通过深度优先搜索(DFS)的方法来解决。我们要做的就是在字符串的每个可能位置插入点&#xff0c;然后检查生成的每一部分是否在 0-255 的范围内&#xff0c;以及是否没有前导零&#xff08;除非这一部分本…...

Mongodb空间索引的使用以及与Django的对接

Mongodb的空间索引 Mongodb数据库大家都非常熟悉&#xff0c;是一个基于分布式文件存储的开源数据库系统&#xff0c;在高负载的情况下&#xff0c;添加更多的节点&#xff0c;可以保证服务器性能&#xff0c;数据结构由键值(key>value)对组成。MongoDB 文档类似于 JSON 对…...

Windows安装MySQL数据库

MySQL数据库安装 MySQL下载 下载地址&#xff1a;https://dev.mysql.com/downloads/mysql/ 可以选择下载msi或zip&#xff0c;以下为zip模式安装步骤 下载了mysql的zip安装包之后解压即可&#xff1b; Windows安装步骤 初始化MySQL&#xff0c;并记录生成的用户密码root的随机…...

聊聊函数式编程中的“式”

当谈到函数式编程的“式”时&#xff0c;通常指的是函数的组合、转换和应用&#xff0c;以及处理数据的方式和风格。在函数式编程中&#xff0c;式是用来构建程序逻辑的基本单元。 下面更详细解释函数式编程中的几个关键式&#xff1a; 函数的组合&#xff1a; 函数式编程中…...

ubuntu目录分析

在Ubuntu根目录下&#xff0c;以下是一些常见文件夹的含义&#xff1a; /bin&#xff1a;存放可执行文件&#xff0c;包含一些基本的命令和工具。 /boot&#xff1a;存放启动时所需的文件&#xff0c;如内核和引导加载程序。 /dev&#xff1a;包含设备文件&#xff0c;用于与硬…...

Python 进阶(三):正则表达式(re 模块)

❤️ 博客主页&#xff1a;水滴技术 &#x1f338; 订阅专栏&#xff1a;Python 入门核心技术 &#x1f680; 支持水滴&#xff1a;点赞&#x1f44d; 收藏⭐ 留言&#x1f4ac; 文章目录 1. 导入re模块2. re模块中的常用函数2.1 re.search()2.2 re.findall()2.3 re.sub()2.4…...

Vue2 第六节 key的作用与原理

&#xff08;1&#xff09;虚拟DOM &#xff08;2&#xff09;v-for中的key的作用 一.虚拟DOM 1.虚拟DOM就是内存中的数据 2.原生的JS没有虚拟DOM: 如果新的数据和原来的数据有重复数据&#xff0c;不会在原来的基础上新加数据&#xff0c;而是重新生成一份 3. Vue会有虚拟…...

黑客为什么不攻击微信钱包?

黑客为什么不攻击微信钱包&#xff1f; 现在人人手机里都装着微信和支付宝&#xff0c;里面都或多或少存了些钱。怎么从来没听说谁的钱被技术牛逼黑客惦记走&#xff1f; 是黑客没攻击过&#xff1f;还是黑客不敢攻击&#xff1f;其实都不是。阿里巴巴首席风险官郑俊芳就说过&…...

5分钟彻底告别风扇噪音!FanControl终极静音配置完全指南

5分钟彻底告别风扇噪音&#xff01;FanControl终极静音配置完全指南 【免费下载链接】FanControl.Releases This is the release repository for Fan Control, a highly customizable fan controlling software for Windows. 项目地址: https://gitcode.com/GitHub_Trending/…...

嵌入式C语言状态机编程实践与优化

1. 状态机编程基础概念在嵌入式系统开发中&#xff0c;状态机(State Machine)是一种极其重要的编程范式。它通过定义系统可能处于的状态集合、状态之间的转换条件以及状态转换时执行的动作&#xff0c;来清晰地描述系统的行为逻辑。状态机之所以在嵌入式领域广泛应用&#xff0…...

从零到一:AI工程开源资源全栈指南与实战应用

从零到一&#xff1a;AI工程开源资源全栈指南与实战应用 【免费下载链接】aie-book [WIP] Resources for AI engineers. Also contains supporting materials for the book AI Engineering (Chip Huyen, 2025) 项目地址: https://gitcode.com/GitHub_Trending/ai/aie-book …...

开箱即用!Qwen-Image-2512-SDNQ Web服务快速体验指南

开箱即用&#xff01;Qwen-Image-2512-SDNQ Web服务快速体验指南 1. 五分钟了解Qwen-Image-2512-SDNQ Web服务 你是否遇到过这样的场景&#xff1a;需要快速生成一张概念图&#xff0c;但打开专业设计软件太麻烦&#xff1f;或者想尝试AI绘画&#xff0c;却被复杂的模型部署步…...

像素剧本圣殿新手指南:RPG对话框系统理解AI输出逻辑与修改技巧

像素剧本圣殿新手指南&#xff1a;RPG对话框系统理解AI输出逻辑与修改技巧 1. 认识像素剧本圣殿的RPG对话框系统 像素剧本圣殿的RPG对话框系统是其最具特色的交互界面&#xff0c;它模拟了经典像素游戏中NPC对话的场景。这个系统不仅仅是视觉上的复古设计&#xff0c;更是AI剧…...

QGC二次开发---多机协同任务中的智能框选与指令批量下发

1. 多机协同作业的核心痛点与解决方案 在农业植保、物流配送等需要多架无人机协同作业的场景中&#xff0c;操作人员经常面临一个棘手问题&#xff1a;如何快速选择特定区域的无人机并批量下发指令&#xff1f;传统方法需要逐个点击无人机图标&#xff0c;效率低下且容易出错。…...

突破方舟生存进化技术壁垒的智能管理工具

突破方舟生存进化技术壁垒的智能管理工具 【免费下载链接】TEKLauncher Launcher for ARK: Survival Evolved 项目地址: https://gitcode.com/gh_mirrors/te/TEKLauncher 你是否曾因MOD安装顺序错误导致游戏频繁崩溃&#xff1f;是否在搭建私人服务器时被端口配置弄得晕…...

【Matlab】MATLAB教程:图形属性修改(案例:set(h,‘Color‘,‘red‘),应用:自定义图形样式)

MATLAB教程:图形属性修改(案例:set(h,Color,red),应用:自定义图形样式) 在MATLAB数据可视化、实验报告绘图、工程结果展示等场景中,默认绘制的图形往往难以满足个性化需求和规范要求。无论是调整线条颜色、粗细,还是优化坐标轴、图例样式,核心目标都是通过图形属性修…...

Linux信号机制:原理、处理与实践

1. Linux信号机制基础解析在Linux系统中&#xff0c;信号是一种进程间通信的重要机制。想象一下你正在厨房做饭&#xff0c;突然门铃响了——这个门铃就相当于Linux系统中的信号&#xff0c;它打断了你当前的工作流程&#xff0c;迫使你做出响应。信号本质上是一种异步事件通知…...