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

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消费积压 如果生产者发送消息的速度过快&#xff0c;或者是消费者处理消息的速度太慢&#xff0c;那么就会有越来越多的消息无法及时消费&#xff0c;也就是消费积压。 消费积压时&#xff0c; (1) 可以增加Topic的分区数&#xff0c;并且增加消费组的消费者数量&#…...

android通过广播打印RAM信息

通过广播打印ram相关log 参数说明&#xff1a; 广播&#xff1a;com.android.settings.action.RAM_INFO int型参数index&#xff1a;0 - 3h, 1 - 6h, 2 - 12h, 3 - 24h 代表过去时间app使用ram情况&#xff08;平均/最大占用&#xff09; Index: frameworks/base/services/cor…...

C++新经典模板与泛型编程:策略类模板

策略类模板 在前面的博文中&#xff0c;策略类SumPolicy和MinPolicy都是普通的类&#xff0c;其中包含的是一个静态成员函数模板algorithm()&#xff0c;该函数模板包含两个类型模板参数。其实&#xff0c;也可以把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替代工具

人工智能技术的发展不仅改变了我们的生活方式&#xff0c;也在各行各业发挥着越来越重要的作用。ChatGPT&#xff08;Generative Pre-trained Transformer&#xff09;作为一种先进的自然语言处理模型&#xff0c;由OpenAI推出&#xff0c;其在生成人类般流畅对话方面表现出色。…...

服务器如何保证数据安全_Maizyun

服务器如何保证数据安全 在当今的数字化时代&#xff0c;数据安全已经成为企业和社会组织必须面对的重要问题。服务器作为存储和处理大量数据的核心组件&#xff0c;必须采取有效的措施来确保数据的安全。本文将探讨服务器如何保证数据安全。 一、访问控制和身份认证 访问控…...

sql2005日志文件过大如何清理

由于安装的时候没有计划好空间&#xff0c;默认装在系统盘&#xff0c;而且又没有做自动备份、截断事务日志等&#xff0c;很快LDF文件就达到十几G&#xff0c;或者几十G &#xff0c;此时就不得不处理了。 备份和计划就不说了&#xff0c;现在就说下怎么把它先删除吧&#xf…...

Linux--学习记录(2)

解压命令&#xff1a; gzip命令&#xff1a; 参数&#xff1a; -k&#xff1a;待压缩的文件会保留下来&#xff0c;生成一个新的压缩文件-d&#xff1a;解压压缩文件语法&#xff1a; gzip -k pathname(待压缩的文件夹名)gzip -kd name.gz&#xff08;待解压的压缩包名&#x…...

字符串函数`strlen`、`strcpy`、`strcmp`、`strstr`、`strcat`的使用以及模拟实现

文章目录 &#x1f680;前言&#x1f680;库函数strlen✈️strlen的模拟实现 &#x1f680;库函数strcpy✈️strcpy的模拟实现 &#x1f680;strcmp✈️strcmp的模拟实现 &#x1f680;strstr✈️strstr的模拟实现 &#x1f680;strcat✈️strcat的模拟实现 &#x1f680;前言 …...

插入排序与希尔排序(C语言实现)

1.插入排序 由上面的动图可以知道插入排序的逻辑就是从第一个元素开始往后遍历&#xff0c;如果找到比前一个元素小的&#xff08;或者大的&#xff09;就往前排&#xff0c;所以插入排序的每一次遍历都会保证前面的数据是有序的&#xff0c;接下类用代码进行讲解。 我们这里传…...

【微软技术栈】与其他.NET语言的互操作性 (C++/CLI)

本文内容 使用 C# 索引器实现 C# 的 is 和 as 关键字实现 C# 的 lock 关键字 本节中的主题介绍如何在 Visual C 中创建程序集&#xff0c;这些程序集使用或提供以 C# 或 Visual Basic 编写的程序集的功能。 1、使用 C# 索引器 Visual C 不包含索引器&#xff1b;它具有索引…...

TCPUDP使用场景讨论

将链路从TCP改为UDP会对通信链路产生以下影响和注意事项&#xff1a; 可靠性&#xff1a;UDP是无连接的协议&#xff0c;与TCP相比&#xff0c;它不提供可靠性保证和重传机制。因此&#xff0c;当将链路从TCP改为UDP时&#xff0c;通信的可靠性会降低。如果在通信过程中丢失了U…...

C#最小二乘法线性回归

文章目录 SimpleRegressionMultipleRegression MathNet系列&#xff1a;矩阵生成 \quad 矩阵计算 LinearRegression是MathNet的线性回归模块&#xff0c;主要包括SimpleRegression和MultipleRegression这两个静态类&#xff0c;前者提供了最小二乘法的线性拟合&#xff0c;后…...

ULAM公链第九十六期工作总结

迈入12月&#xff0c;接下来就是雪花&#xff0c;圣诞&#xff0c;新年和更好的我们&#xff01;愿生活不拥挤&#xff0c;笑容不必刻意&#xff0c;愿一切美好如期而至&#xff01; 2023年11月01日—2023年12月01日关于ULAM这期工作汇报&#xff0c;我们通过技术板块&#xff…...

基于Echarts的大数据可视化模板:智慧交通管理

目录 引言智慧交通管理的重要性ECharts在智慧交通中的作用智慧交通管理系统架构系统总体架构数据收集与处理Echarts与大数据可视化Echarts库以及其在大数据可视化领域的应用优势开发过程和所选设计方案模板如何满足管理的特定需求模板功能与特性深入解析模板提供的各项功能模板…...

C#-快速剖析文件和流,并使用

目录 一、概述 二、文件系统 1、检查驱动器信息 2、Path 3、文件和文件夹 三、流 1、FileStream 2、StreamWriter与StreamReader 3、BinaryWriter与BinaryReader 一、概述 文件&#xff0c;具有永久存储及特定顺序的字节组成的一个有序、具有名称的集合&#xff1b; …...

【Linux】如何在Ubuntu 20.04上安装PostgreSQL

介绍 PostgreSQL或Postgres是一个关系数据库管理系统&#xff0c;提供SQL查询语言的实现。它符合标准&#xff0c;具有许多高级功能&#xff0c;如可靠的事务和无读锁的并发性。 本指南演示了如何在Ubuntu 20.04服务器上快速启动和运行Postgres&#xff0c;从安装PostgreSQL到…...

IT程序员面试题目汇总及答案-计算机面试

程序员面试题目汇总及答案-计算机面试 问题1:请你描述一下你在过去的工作中遇到的一个技术难题,你是如何解决的? 答案1:在我之前的工作中,我遇到了一个涉及大数据处理的问题。由于数据量巨大,传统的处理方法无法在规定的时间内完成。我最后采用了一种分布式计算的方法,…...

IGP(Interior Gateway Protocol,内部网关协议)

IGP&#xff08;Interior Gateway Protocol&#xff0c;内部网关协议&#xff09; 是一种用于在一个自治系统&#xff08;AS&#xff09;内部传递路由信息的路由协议&#xff0c;主要用于在一个组织或机构的内部网络中决定数据包的最佳路径。与用于自治系统之间通信的 EGP&…...

如何在网页里填写 PDF 表格?

有时候&#xff0c;你可能希望用户能在你的网站上填写 PDF 表单。然而&#xff0c;这件事并不简单&#xff0c;因为 PDF 并不是一种原生的网页格式。虽然浏览器可以显示 PDF 文件&#xff0c;但原生并不支持编辑或填写它们。更糟的是&#xff0c;如果你想收集表单数据&#xff…...

Unsafe Fileupload篇补充-木马的详细教程与木马分享(中国蚁剑方式)

在之前的皮卡丘靶场第九期Unsafe Fileupload篇中我们学习了木马的原理并且学了一个简单的木马文件 本期内容是为了更好的为大家解释木马&#xff08;服务器方面的&#xff09;的原理&#xff0c;连接&#xff0c;以及各种木马及连接工具的分享 文件木马&#xff1a;https://w…...

CVE-2020-17519源码分析与漏洞复现(Flink 任意文件读取)

漏洞概览 漏洞名称&#xff1a;Apache Flink REST API 任意文件读取漏洞CVE编号&#xff1a;CVE-2020-17519CVSS评分&#xff1a;7.5影响版本&#xff1a;Apache Flink 1.11.0、1.11.1、1.11.2修复版本&#xff1a;≥ 1.11.3 或 ≥ 1.12.0漏洞类型&#xff1a;路径遍历&#x…...

【网络安全】开源系统getshell漏洞挖掘

审计过程&#xff1a; 在入口文件admin/index.php中&#xff1a; 用户可以通过m,c,a等参数控制加载的文件和方法&#xff0c;在app/system/entrance.php中存在重点代码&#xff1a; 当M_TYPE system并且M_MODULE include时&#xff0c;会设置常量PATH_OWN_FILE为PATH_APP.M_T…...

关于uniapp展示PDF的解决方案

在 UniApp 的 H5 环境中使用 pdf-vue3 组件可以实现完整的 PDF 预览功能。以下是详细实现步骤和注意事项&#xff1a; 一、安装依赖 安装 pdf-vue3 和 PDF.js 核心库&#xff1a; npm install pdf-vue3 pdfjs-dist二、基本使用示例 <template><view class"con…...

在鸿蒙HarmonyOS 5中使用DevEco Studio实现指南针功能

指南针功能是许多位置服务应用的基础功能之一。下面我将详细介绍如何在HarmonyOS 5中使用DevEco Studio实现指南针功能。 1. 开发环境准备 确保已安装DevEco Studio 3.1或更高版本确保项目使用的是HarmonyOS 5.0 SDK在项目的module.json5中配置必要的权限 2. 权限配置 在mo…...

6.9-QT模拟计算器

源码: 头文件: widget.h #ifndef WIDGET_H #define WIDGET_H#include <QWidget> #include <QMouseEvent>QT_BEGIN_NAMESPACE namespace Ui { class Widget; } QT_END_NAMESPACEclass Widget : public QWidget {Q_OBJECTpublic:Widget(QWidget *parent nullptr);…...

数据结构第5章:树和二叉树完全指南(自整理详细图文笔记)

名人说&#xff1a;莫道桑榆晚&#xff0c;为霞尚满天。——刘禹锡&#xff08;刘梦得&#xff0c;诗豪&#xff09; 原创笔记&#xff1a;Code_流苏(CSDN)&#xff08;一个喜欢古诗词和编程的Coder&#x1f60a;&#xff09; 上一篇&#xff1a;《数据结构第4章 数组和广义表》…...

GraphRAG优化新思路-开源的ROGRAG框架

目前的如微软开源的GraphRAG的工作流程都较为复杂&#xff0c;难以孤立地评估各个组件的贡献&#xff0c;传统的检索方法在处理复杂推理任务时可能不够有效&#xff0c;特别是在需要理解实体间关系或多跳知识的情况下。先说结论&#xff0c;看完后感觉这个框架性能上不会比Grap…...