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实现登录拦截
如果我们不进行登录拦截的话,即使我们跳过登录页面直接去访问任意一个页面也能访问成功,那么登录功能就没有意义,同时也会存在安全问题,因为有些操作是要用户登录后才能执行的,如果用户没有登录,该接口就获…...
浅谈泛在电力物联网、能源互联网与虚拟电厂
导读:从能源互联网推进受阻,到泛在电力物联网名噪一时,到虚拟电厂再次走向火爆,能源领域亟需更进一步的数智化发展。如今,随着新型电力系统建设推进,虚拟电厂有望迎来快速发展。除了国网和南网公司下属的电…...
《Qt C++ 与 OpenCV:解锁视频播放程序设计的奥秘》
引言:探索视频播放程序设计之旅 在当今数字化时代,多媒体应用已渗透到我们生活的方方面面,从日常的视频娱乐到专业的视频监控、视频会议系统,视频播放程序作为多媒体应用的核心组成部分,扮演着至关重要的角色。无论是在个人电脑、移动设备还是智能电视等平台上,用户都期望…...
聊聊 Pulsar:Producer 源码解析
一、前言 Apache Pulsar 是一个企业级的开源分布式消息传递平台,以其高性能、可扩展性和存储计算分离架构在消息队列和流处理领域独树一帜。在 Pulsar 的核心架构中,Producer(生产者) 是连接客户端应用与消息队列的第一步。生产者…...
JVM垃圾回收机制全解析
Java虚拟机(JVM)中的垃圾收集器(Garbage Collector,简称GC)是用于自动管理内存的机制。它负责识别和清除不再被程序使用的对象,从而释放内存空间,避免内存泄漏和内存溢出等问题。垃圾收集器在Ja…...
linux 下常用变更-8
1、删除普通用户 查询用户初始UID和GIDls -l /home/ ###家目录中查看UID cat /etc/group ###此文件查看GID删除用户1.编辑文件 /etc/passwd 找到对应的行,YW343:x:0:0::/home/YW343:/bin/bash 2.将标红的位置修改为用户对应初始UID和GID: YW3…...
DBAPI如何优雅的获取单条数据
API如何优雅的获取单条数据 案例一 对于查询类API,查询的是单条数据,比如根据主键ID查询用户信息,sql如下: select id, name, age from user where id #{id}API默认返回的数据格式是多条的,如下: {&qu…...
鱼香ros docker配置镜像报错:https://registry-1.docker.io/v2/
使用鱼香ros一件安装docker时的https://registry-1.docker.io/v2/问题 一键安装指令 wget http://fishros.com/install -O fishros && . fishros出现问题:docker pull 失败 网络不同,需要使用镜像源 按照如下步骤操作 sudo vi /etc/docker/dae…...
零基础设计模式——行为型模式 - 责任链模式
第四部分:行为型模式 - 责任链模式 (Chain of Responsibility Pattern) 欢迎来到行为型模式的学习!行为型模式关注对象之间的职责分配、算法封装和对象间的交互。我们将学习的第一个行为型模式是责任链模式。 核心思想:使多个对象都有机会处…...
CRMEB 框架中 PHP 上传扩展开发:涵盖本地上传及阿里云 OSS、腾讯云 COS、七牛云
目前已有本地上传、阿里云OSS上传、腾讯云COS上传、七牛云上传扩展 扩展入口文件 文件目录 crmeb\services\upload\Upload.php namespace crmeb\services\upload;use crmeb\basic\BaseManager; use think\facade\Config;/*** Class Upload* package crmeb\services\upload* …...
IoT/HCIP实验-3/LiteOS操作系统内核实验(任务、内存、信号量、CMSIS..)
文章目录 概述HelloWorld 工程C/C配置编译器主配置Makefile脚本烧录器主配置运行结果程序调用栈 任务管理实验实验结果osal 系统适配层osal_task_create 其他实验实验源码内存管理实验互斥锁实验信号量实验 CMISIS接口实验还是得JlINKCMSIS 简介LiteOS->CMSIS任务间消息交互…...
【开发技术】.Net使用FFmpeg视频特定帧上绘制内容
目录 一、目的 二、解决方案 2.1 什么是FFmpeg 2.2 FFmpeg主要功能 2.3 使用Xabe.FFmpeg调用FFmpeg功能 2.4 使用 FFmpeg 的 drawbox 滤镜来绘制 ROI 三、总结 一、目的 当前市场上有很多目标检测智能识别的相关算法,当前调用一个医疗行业的AI识别算法后返回…...
