kafka入门(三):kafka多线程消费
kafka消费积压
如果生产者发送消息的速度过快,或者是消费者处理消息的速度太慢,那么就会有越来越多的消息无法及时消费,也就是消费积压。
消费积压时,
(1) 可以增加Topic的分区数,并且增加消费组的消费者数量,让消费者数等于分区数。
(2) 还可以使用多线程消费,提高消费速度。
kafka多线程消费的代码:
public class ThirdMultiConsumerThreadDemo {public static final String BROKER_LIST = "localhost:9092";public static final String TOPIC = "myTopic1";public static final String GROUP_ID = "group.demo";public static void main(String[] args) {Properties props = initConfig();KafkaConsumerThread consumerThread = new KafkaConsumerThread(props, TOPIC,Runtime.getRuntime().availableProcessors());consumerThread.start();}/**** kafka配置* @return*/public static Properties initConfig() {Properties props = new Properties();props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);return props;}/*** kafka消费者线程*/public static class KafkaConsumerThread extends Thread {private KafkaConsumer<String, String> kafkaConsumer;private ExecutorService executorService;private int threadNumber;public KafkaConsumerThread(Properties props, String topic, int threadNumber) {kafkaConsumer = new KafkaConsumer<>(props);kafkaConsumer.subscribe(Collections.singletonList(topic));this.threadNumber = threadNumber;executorService = new ThreadPoolExecutor(threadNumber, threadNumber,0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(1000),new ThreadPoolExecutor.CallerRunsPolicy());}@Overridepublic void run() {try {while (true) {ConsumerRecords<String, String> records =kafkaConsumer.poll(Duration.ofMillis(100));if (!records.isEmpty()) {executorService.submit(new RecordsHandler(records));}}} catch (Exception e) {log.error("run error", e);} finally {kafkaConsumer.close();}}}/*** 处理消息*/public static class RecordsHandler extends Thread {public final ConsumerRecords<String, String> records;public RecordsHandler(ConsumerRecords<String, String> records) {this.records = records;}@Overridepublic void run() {//处理records.for (ConsumerRecord<String, String> record : records) {System.out.println("==========>record:"+record.value() + ",thread:" + Thread.currentThread().getName());}}}}
发送消息后,使用多线程消息,运行结果如下:
==========>record:{"id":"1234","name":"lin"},thread:pool-1-thread-1
==========>record:{"id":"5678","name":"chen"},thread:pool-1-thread-2
==========>record:{"id":"91011","name":"wu"},thread:pool-1-thread-3
参考资料:
《深入理解Kafka:核心设计与实践原理》
相关文章:
kafka入门(三):kafka多线程消费
kafka消费积压 如果生产者发送消息的速度过快,或者是消费者处理消息的速度太慢,那么就会有越来越多的消息无法及时消费,也就是消费积压。 消费积压时, (1) 可以增加Topic的分区数,并且增加消费组的消费者数量&#…...
android通过广播打印RAM信息
通过广播打印ram相关log 参数说明: 广播:com.android.settings.action.RAM_INFO int型参数index:0 - 3h, 1 - 6h, 2 - 12h, 3 - 24h 代表过去时间app使用ram情况(平均/最大占用) Index: frameworks/base/services/cor…...
C++新经典模板与泛型编程:策略类模板
策略类模板 在前面的博文中,策略类SumPolicy和MinPolicy都是普通的类,其中包含的是一个静态成员函数模板algorithm(),该函数模板包含两个类型模板参数。其实,也可以把SumPolicy和MinPolicy类写成类模板—直接把algorithm()中的两…...
微信小程序引入Vant Weapp修改样式不起作用,使用外部样式类进行覆盖
一、引入Vant Weapp后样式问题 在项目中使用第三方组件修改css样式时,总是出现各种各样问题,修改的css样式不起作用,没有效果,效果不符合预期等。 栗子(引入一个搜索框组件)实现效果: 左侧有一个搜索文字背景为蓝色,接着跟一个搜索框 wxml <view class"container&q…...
python核酸检测 青少年电子学会等级考试 中小学生python编程等级考试二级真题答案解析2022年6月
目录 python核酸检测 一、题目要求 1、编程实现 2、输入输出...
搭建React项目,基于Vite+React+TS+ESLint+Prettier+Husky+Commitlint
基于ViteReactTSESLintPrettierHuskyCommitlint搭建React项目 node: 20.10.0 一、创建项目 安装包管理器pnpm npm i pnpm -g基于Vite创建项目 pnpm create vitelatest web-gis-react --template react-ts进入项目目录安装依赖 $ cd web-gis-react $ pnpm i启动项目 $ pnpm…...
ChatGPT在国内的使用限制,国内的ChatGPT替代工具
人工智能技术的发展不仅改变了我们的生活方式,也在各行各业发挥着越来越重要的作用。ChatGPT(Generative Pre-trained Transformer)作为一种先进的自然语言处理模型,由OpenAI推出,其在生成人类般流畅对话方面表现出色。…...
服务器如何保证数据安全_Maizyun
服务器如何保证数据安全 在当今的数字化时代,数据安全已经成为企业和社会组织必须面对的重要问题。服务器作为存储和处理大量数据的核心组件,必须采取有效的措施来确保数据的安全。本文将探讨服务器如何保证数据安全。 一、访问控制和身份认证 访问控…...
sql2005日志文件过大如何清理
由于安装的时候没有计划好空间,默认装在系统盘,而且又没有做自动备份、截断事务日志等,很快LDF文件就达到十几G,或者几十G ,此时就不得不处理了。 备份和计划就不说了,现在就说下怎么把它先删除吧…...
Linux--学习记录(2)
解压命令: gzip命令: 参数: -k:待压缩的文件会保留下来,生成一个新的压缩文件-d:解压压缩文件语法: gzip -k pathname(待压缩的文件夹名)gzip -kd name.gz(待解压的压缩包名&#x…...
字符串函数`strlen`、`strcpy`、`strcmp`、`strstr`、`strcat`的使用以及模拟实现
文章目录 🚀前言🚀库函数strlen✈️strlen的模拟实现 🚀库函数strcpy✈️strcpy的模拟实现 🚀strcmp✈️strcmp的模拟实现 🚀strstr✈️strstr的模拟实现 🚀strcat✈️strcat的模拟实现 🚀前言 …...
插入排序与希尔排序(C语言实现)
1.插入排序 由上面的动图可以知道插入排序的逻辑就是从第一个元素开始往后遍历,如果找到比前一个元素小的(或者大的)就往前排,所以插入排序的每一次遍历都会保证前面的数据是有序的,接下类用代码进行讲解。 我们这里传…...
【微软技术栈】与其他.NET语言的互操作性 (C++/CLI)
本文内容 使用 C# 索引器实现 C# 的 is 和 as 关键字实现 C# 的 lock 关键字 本节中的主题介绍如何在 Visual C 中创建程序集,这些程序集使用或提供以 C# 或 Visual Basic 编写的程序集的功能。 1、使用 C# 索引器 Visual C 不包含索引器;它具有索引…...
TCPUDP使用场景讨论
将链路从TCP改为UDP会对通信链路产生以下影响和注意事项: 可靠性:UDP是无连接的协议,与TCP相比,它不提供可靠性保证和重传机制。因此,当将链路从TCP改为UDP时,通信的可靠性会降低。如果在通信过程中丢失了U…...
C#最小二乘法线性回归
文章目录 SimpleRegressionMultipleRegression MathNet系列:矩阵生成 \quad 矩阵计算 LinearRegression是MathNet的线性回归模块,主要包括SimpleRegression和MultipleRegression这两个静态类,前者提供了最小二乘法的线性拟合,后…...
ULAM公链第九十六期工作总结
迈入12月,接下来就是雪花,圣诞,新年和更好的我们!愿生活不拥挤,笑容不必刻意,愿一切美好如期而至! 2023年11月01日—2023年12月01日关于ULAM这期工作汇报,我们通过技术板块ÿ…...
基于Echarts的大数据可视化模板:智慧交通管理
目录 引言智慧交通管理的重要性ECharts在智慧交通中的作用智慧交通管理系统架构系统总体架构数据收集与处理Echarts与大数据可视化Echarts库以及其在大数据可视化领域的应用优势开发过程和所选设计方案模板如何满足管理的特定需求模板功能与特性深入解析模板提供的各项功能模板…...
C#-快速剖析文件和流,并使用
目录 一、概述 二、文件系统 1、检查驱动器信息 2、Path 3、文件和文件夹 三、流 1、FileStream 2、StreamWriter与StreamReader 3、BinaryWriter与BinaryReader 一、概述 文件,具有永久存储及特定顺序的字节组成的一个有序、具有名称的集合; …...
【Linux】如何在Ubuntu 20.04上安装PostgreSQL
介绍 PostgreSQL或Postgres是一个关系数据库管理系统,提供SQL查询语言的实现。它符合标准,具有许多高级功能,如可靠的事务和无读锁的并发性。 本指南演示了如何在Ubuntu 20.04服务器上快速启动和运行Postgres,从安装PostgreSQL到…...
IT程序员面试题目汇总及答案-计算机面试
程序员面试题目汇总及答案-计算机面试 问题1:请你描述一下你在过去的工作中遇到的一个技术难题,你是如何解决的? 答案1:在我之前的工作中,我遇到了一个涉及大数据处理的问题。由于数据量巨大,传统的处理方法无法在规定的时间内完成。我最后采用了一种分布式计算的方法,…...
多云管理“拦路虎”:深入解析网络互联、身份同步与成本可视化的技术复杂度
一、引言:多云环境的技术复杂性本质 企业采用多云策略已从技术选型升维至生存刚需。当业务系统分散部署在多个云平台时,基础设施的技术债呈现指数级积累。网络连接、身份认证、成本管理这三大核心挑战相互嵌套:跨云网络构建数据…...
synchronized 学习
学习源: https://www.bilibili.com/video/BV1aJ411V763?spm_id_from333.788.videopod.episodes&vd_source32e1c41a9370911ab06d12fbc36c4ebc 1.应用场景 不超卖,也要考虑性能问题(场景) 2.常见面试问题: sync出…...
工程地质软件市场:发展现状、趋势与策略建议
一、引言 在工程建设领域,准确把握地质条件是确保项目顺利推进和安全运营的关键。工程地质软件作为处理、分析、模拟和展示工程地质数据的重要工具,正发挥着日益重要的作用。它凭借强大的数据处理能力、三维建模功能、空间分析工具和可视化展示手段&…...
linux arm系统烧录
1、打开瑞芯微程序 2、按住linux arm 的 recover按键 插入电源 3、当瑞芯微检测到有设备 4、松开recover按键 5、选择升级固件 6、点击固件选择本地刷机的linux arm 镜像 7、点击升级 (忘了有没有这步了 估计有) 刷机程序 和 镜像 就不提供了。要刷的时…...
Neo4j 集群管理:原理、技术与最佳实践深度解析
Neo4j 的集群技术是其企业级高可用性、可扩展性和容错能力的核心。通过深入分析官方文档,本文将系统阐述其集群管理的核心原理、关键技术、实用技巧和行业最佳实践。 Neo4j 的 Causal Clustering 架构提供了一个强大而灵活的基石,用于构建高可用、可扩展且一致的图数据库服务…...
Android Bitmap治理全解析:从加载优化到泄漏防控的全生命周期管理
引言 Bitmap(位图)是Android应用内存占用的“头号杀手”。一张1080P(1920x1080)的图片以ARGB_8888格式加载时,内存占用高达8MB(192010804字节)。据统计,超过60%的应用OOM崩溃与Bitm…...
初学 pytest 记录
安装 pip install pytest用例可以是函数也可以是类中的方法 def test_func():print()class TestAdd: # def __init__(self): 在 pytest 中不可以使用__init__方法 # self.cc 12345 pytest.mark.api def test_str(self):res add(1, 2)assert res 12def test_int(self):r…...
jmeter聚合报告中参数详解
sample、average、min、max、90%line、95%line,99%line、Error错误率、吞吐量Thoughput、KB/sec每秒传输的数据量 sample(样本数) 表示测试中发送的请求数量,即测试执行了多少次请求。 单位,以个或者次数表示。 示例:…...
从 GreenPlum 到镜舟数据库:杭银消费金融湖仓一体转型实践
作者:吴岐诗,杭银消费金融大数据应用开发工程师 本文整理自杭银消费金融大数据应用开发工程师在StarRocks Summit Asia 2024的分享 引言:融合数据湖与数仓的创新之路 在数字金融时代,数据已成为金融机构的核心竞争力。杭银消费金…...
人工智能--安全大模型训练计划:基于Fine-tuning + LLM Agent
安全大模型训练计划:基于Fine-tuning LLM Agent 1. 构建高质量安全数据集 目标:为安全大模型创建高质量、去偏、符合伦理的训练数据集,涵盖安全相关任务(如有害内容检测、隐私保护、道德推理等)。 1.1 数据收集 描…...
