Kafka3.0.0版本——消费者(独立消费者消费某一个主题中某个分区数据案例__订阅分区)
目录
- 一、独立消费者消费某一个主题中某个分区数据案例
- 1.1、案例需求
- 1.2、案例代码
- 1.3、测试
一、独立消费者消费某一个主题中某个分区数据案例
1.1、案例需求
- 创建一个独立消费者,消费firstTopic主题 0 号分区的数据,所下图所示:

1.2、案例代码
-
生产者往firstTopic主题 0 号分区发送数据代码
package com.xz.kafka.producer;import org.apache.kafka.clients.producer.*; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties;public class CustomProducerCallback {public static void main(String[] args) throws InterruptedException {//1、创建 kafka 生产者的配置对象Properties properties = new Properties();//2、给 kafka 配置对象添加配置信息:bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.136.27:9092,192.168.136.28:9092,192.168.136.29:9092");//3、指定对应的key和value的序列化类型 key.serializer value.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());//4、创建 kafka 生产者对象KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);//5、调用 send 方法,发送消息for (int i = 0; i < 5; i++) {kafkaProducer.send(new ProducerRecord<>("firstTopic", 0,"","hello kafka" + i), new Callback() {@Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if (exception == null){System.out.println("主题: "+metadata.topic() + " 分区: "+ metadata.partition());}}});Thread.sleep(2);}// 3 关闭资源kafkaProducer.close();} } -
消费者消费firstTopic主题 0 分区数据代码
package com.xz.kafka.consumer;import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration; import java.util.ArrayList; import java.util.Properties;public class CustomConsumerPartition {public static void main(String[] args) {// 配置Properties properties = new Properties();// 连接properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.136.27:9092,192.168.136.28:9092,192.168.136.29:9092");// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");// 1 创建一个消费者KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);// 2 订阅主题对应的分区ArrayList<TopicPartition> topicPartitions = new ArrayList<>();topicPartitions.add(new TopicPartition("firstTopic",0));kafkaConsumer.assign(topicPartitions);// 3 消费数据while (true){ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println(consumerRecord);}}} }
1.3、测试
- 在 IDEA 中执行消费者程序,如下图:

- 在 IDEA 中执行生产者程序 ,在控制台观察生成几个 0号分区的数据,如下图:

- 在 IDEA 控制台,观察接收到的数据,只能消费到 0 号分区数据表示正确。

相关文章:
Kafka3.0.0版本——消费者(独立消费者消费某一个主题中某个分区数据案例__订阅分区)
目录 一、独立消费者消费某一个主题中某个分区数据案例1.1、案例需求1.2、案例代码1.3、测试 一、独立消费者消费某一个主题中某个分区数据案例 1.1、案例需求 创建一个独立消费者,消费firstTopic主题 0 号分区的数据,所下图所示: 1.2、案…...
基于Simulink的用于电力系统动态分析
💥💥💞💞欢迎来到本博客❤️❤️💥💥 🏆博主优势:🌞🌞🌞博客内容尽量做到思维缜密,逻辑清晰,为了方便读者。 ⛳️座右铭&a…...
日200亿次调用,喜马拉雅网关的架构设计
说在前面 在40岁老架构师 尼恩的读者社区(50)中,很多小伙伴拿到一线互联网企业如阿里、网易、有赞、希音、百度、滴滴的面试资格。 最近,尼恩指导一个小伙伴简历,写了一个《API网关项目》,此项目帮这个小伙拿到 字节/阿里/微博/…...
构造函数和析构函数(个人学习笔记黑马学习)
构造函数:主要作用在于创建对象时为对象的成员属性赋值,构造函数由编译器自动调用,无须手动调用。析构函数:主要作用在于对象销毁前系统自动调用,执行一些清理工作。 #include <iostream> using namespace std;//对象初始化和清理class…...
GPT引领前沿与应用突破之GPT4科研实践技术与AI绘图教程
详情点击链接:GPT引领前沿与应用突破之GPT4科研实践技术与AI绘图教程 前沿 GPT对于每个科研人员已经成为不可或缺的辅助工具,不同的研究领域和项目具有不同的需求。 如在科研编程、绘图领域: 1、编程建议和示例代码: 无论你使用的编程语言是…...
Git上传新项目
第一步:初始化 Git 仓库 首先,打开终端或命令行界面,然后导航到项目目录。运行下面的命令来初始化一个新的 Git 仓库: git init这将创建一个新的 .git 子目录,其中包含了初始化的 Git 仓库。 第二步:添加…...
C语言文件操作总结
目录 字符方式读入文件 数据块方式读写文件 文件定位与随机读写 文件中数据的修改 字符方式读入文件 1.向文件中写入(输入字符) 用 fputc 函数或 puts 函数可以把一个字符写到磁盘文件中去。 int fputc(int ch,FILE * fp) ch 是要输出的字符&#…...
原生js之dom如何进行事件监听(事件捕获/冒泡)
那么好,这次主要讲解的就是dom是如何进行事件监听和事件取消监听的,我们知道vue中主要用watch来进行监听. js监听与取消监听 那么原生js主要用到的就是addListenEvent事件来进行监听,可以监听文档dom对象也可以监听浏览器bom对象,监听事件的语法结构如下 Dom/Bom监听 eleme…...
使用SimPowerSystems并网光伏阵列研究(Simulink实现)
💥💥💞💞欢迎来到本博客❤️❤️💥💥 🏆博主优势:🌞🌞🌞博客内容尽量做到思维缜密,逻辑清晰,为了方便读者。 ⛳️座右铭&a…...
BUUCTF-WEB-[ACTF2020 新生赛]Includel
打开靶机 点击tips 利用Burp抓包,未见异常 但发现了响应头是 PHP/7.3.13 想到了"php://input"伪协议POST发送PHP代码 构建Payload:?filephp://filter/readconvert.base64-encode/resourceflag.php 这里需要注意的是使用php://filter伪协议…...
算法通关村十四关:白银挑战-堆能高效解决的经典问题
白银挑战-堆能高效解决的经典问题 1.在数组中找第K大的元素 LeetCode215 https://leetcode.cn/problems/kth-largest-element-in-an-array/ 思路分析 主要解决方法有3个,选择法,堆查找法和快速排序法 方法1:选择法 先遍历一遍找到最大的…...
跨站请求伪造(CSRF)攻击与防御原理
跨站请求伪造(CSRF) 1.1 CSRF原理 1.1.1 基本概念 跨站请求伪造(Cross Site Request Forgery,CSRF)是一种攻击,它强制浏览器客户端用户在当前对其进行身份验证后的Web 应用程序上执行非本意操作的攻击&a…...
从0到1实现播放控制器
这系列文章主要讲诉如何从0到1使用QT实现带时间显示、滚动字幕等的自定义配置视频播放控制器。平时我们乘坐地铁经常看到各条线的播放控制器都大同小异。其实都是通过QT等界面开发软件来实现的。 在具体开发之前,需要明确我们需要做什么? 1. 开发一个可…...
【Vue-Element-Admin】导出el-table全部数据
背景 因为el-table实现了分页查询,所以想要实现el-table需要重新编写一个查询全部数据的方法 查询全部数据 listQuery: export default{return{listQuery:{//page:1,//limit:20,//如果想兼容按条件导出,可以定义查询条件age:undefined,sex:undefined…...
MFC 更改控件的大小和位置
获取当前主窗体的位置rect CRect dlgNow;GetWindowRect(&dlgNow);获取某一个控件当前的位置 CRect rect;CButton* pBtn (CButton*)GetDlgItem(IDC_BUTTONXXX);//获取按钮控件pBtn->GetWindowRect(rect);CWnd* pWnd(CWnd*)GetDlgItem(IDC_EDITXXX);//其它控件࿰…...
【向量数据库】相似向量检索Faiss数据库的安装及余弦相似度计算(C++)
目录 简介安装方法安装OpenBLAS安装lapack编译Faiss 代码示例余弦相似度计算输出ID号而非索引的改进版 简介 Faiss 是一个强大的向量相似度搜索库,具有以下优点: 高效的搜索性能:Faiss 在处理大规模向量数据时表现出色。它利用了高度优化的索…...
教育培训小程序的设计与功能解析
随着互联网的发展,线上教育逐渐成为一种趋势,越来越多的人开始选择在线学习。而搭建一个适合自己的线上教育小程序,可以为教育机构或个人提供更好的教学和学习体验。在本文中,我们将介绍如何通过一个第三方制作平台来搭建在线教育…...
【ES】illegal_argument_exception“,“reason“:“Result window is too large
查询ES数据返回错误: {"root_cause":[{"type":"illegal_argument_exception","reason":"Result window is too large, from size must be less than or equal to: [10000] but was [999999]. See the scroll api for…...
SpringBoot实现登录拦截
如果我们不进行登录拦截的话,即使我们跳过登录页面直接去访问任意一个页面也能访问成功,那么登录功能就没有意义,同时也会存在安全问题,因为有些操作是要用户登录后才能执行的,如果用户没有登录,该接口就获…...
浅谈泛在电力物联网、能源互联网与虚拟电厂
导读:从能源互联网推进受阻,到泛在电力物联网名噪一时,到虚拟电厂再次走向火爆,能源领域亟需更进一步的数智化发展。如今,随着新型电力系统建设推进,虚拟电厂有望迎来快速发展。除了国网和南网公司下属的电…...
挑战杯推荐项目
“人工智能”创意赛 - 智能艺术创作助手:借助大模型技术,开发能根据用户输入的主题、风格等要求,生成绘画、音乐、文学作品等多种形式艺术创作灵感或初稿的应用,帮助艺术家和创意爱好者激发创意、提高创作效率。 - 个性化梦境…...
【网络】每天掌握一个Linux命令 - iftop
在Linux系统中,iftop是网络管理的得力助手,能实时监控网络流量、连接情况等,帮助排查网络异常。接下来从多方面详细介绍它。 目录 【网络】每天掌握一个Linux命令 - iftop工具概述安装方式核心功能基础用法进阶操作实战案例面试题场景生产场景…...
1688商品列表API与其他数据源的对接思路
将1688商品列表API与其他数据源对接时,需结合业务场景设计数据流转链路,重点关注数据格式兼容性、接口调用频率控制及数据一致性维护。以下是具体对接思路及关键技术点: 一、核心对接场景与目标 商品数据同步 场景:将1688商品信息…...
c#开发AI模型对话
AI模型 前面已经介绍了一般AI模型本地部署,直接调用现成的模型数据。这里主要讲述讲接口集成到我们自己的程序中使用方式。 微软提供了ML.NET来开发和使用AI模型,但是目前国内可能使用不多,至少实践例子很少看见。开发训练模型就不介绍了&am…...
【JavaSE】绘图与事件入门学习笔记
-Java绘图坐标体系 坐标体系-介绍 坐标原点位于左上角,以像素为单位。 在Java坐标系中,第一个是x坐标,表示当前位置为水平方向,距离坐标原点x个像素;第二个是y坐标,表示当前位置为垂直方向,距离坐标原点y个像素。 坐标体系-像素 …...
OpenPrompt 和直接对提示词的嵌入向量进行训练有什么区别
OpenPrompt 和直接对提示词的嵌入向量进行训练有什么区别 直接训练提示词嵌入向量的核心区别 您提到的代码: prompt_embedding = initial_embedding.clone().requires_grad_(True) optimizer = torch.optim.Adam([prompt_embedding...
Java数值运算常见陷阱与规避方法
整数除法中的舍入问题 问题现象 当开发者预期进行浮点除法却误用整数除法时,会出现小数部分被截断的情况。典型错误模式如下: void process(int value) {double half = value / 2; // 整数除法导致截断// 使用half变量 }此时...
三分算法与DeepSeek辅助证明是单峰函数
前置 单峰函数有唯一的最大值,最大值左侧的数值严格单调递增,最大值右侧的数值严格单调递减。 单谷函数有唯一的最小值,最小值左侧的数值严格单调递减,最小值右侧的数值严格单调递增。 三分的本质 三分和二分一样都是通过不断缩…...
从物理机到云原生:全面解析计算虚拟化技术的演进与应用
前言:我的虚拟化技术探索之旅 我最早接触"虚拟机"的概念是从Java开始的——JVM(Java Virtual Machine)让"一次编写,到处运行"成为可能。这个软件层面的虚拟化让我着迷,但直到后来接触VMware和Doc…...
C++_哈希表
本篇文章是对C学习的哈希表部分的学习分享 相信一定会对你有所帮助~ 那咱们废话不多说,直接开始吧! 一、基础概念 1. 哈希核心思想: 哈希函数的作用:通过此函数建立一个Key与存储位置之间的映射关系。理想目标:实现…...
