Kafka系列之如何提高消费者消费速度
前言
在实际开发过程中,如果使用Kafka处理超大数据量(千万级、亿级)的场景,Kafka消费者的消费速度可能决定系统性能瓶颈。
实现方案
为了提高消费者的消费速度,我们可以采取以下措施:
- 将主题的分区数量增大,如 20,通过
concurrency将消费者的消费线程数增大到 10(2个pod),提高消息处理的并发能力。 - 将每次批量拉取消息的数量
max.poll.records增大到 500,提高单次处理消息的数量。 - 将消息切分成批次,将单个批次的数据处理业务逻辑放进线程池中异步进行,提高并发处理消息的速度。
- 将异步线程池的拒绝模式调整为 CallerRunsPolicy,这个配置非常重要。当线程池的任务队列已满且所有线程都在忙碌时,新的任务将由提交任务的线程(即调用者线程)来执行。否则在消息量特别大的情况下,很可能会因为线程池任务队列满了而丢失数据。
- 将异步线程池的队列容量设置为 0,这样意味着所有任务必须立即由线程池中的线程来处理,减少在队列中的等待时间。
- 在数据上报的时候进行幂等性验证,防止重复上报数据。
@Component
public class OrderConsumer {@Resource(name = "execThreadPool")private ThreadPoolTaskExecutor execThreadPool;@KafkaListener(id = "record_consumer",topics = "record",groupId = "g_record_consumer",concurrency = "10",properties = {"max.poll.interval.ms:300000", "max.poll.records:500"})public void consume(ConsumerRecords<String, String> records, Acknowledgment ack) {execThreadPool.submit(()-> {// 业务逻辑});ack.acknowledge();}}
ThreadPoolTaskExecutor 是 Spring 框架提供的一个线程池实现,用于管理和执行多线程任务。它是 TaskExecutor 接口的实现,提供了在 Spring 应用程序中创建和配置线程池的便捷方式。
ThreadPoolTaskExecutor主要特点:
-
线程池配置: ThreadPoolTaskExecutor 允许你配置核心线程数、最大线程数、队列容量等线程池属性。
-
线程创建和销毁: 它会根据任务的需求自动创建和销毁线程,避免不必要的线程创建和销毁开销。
-
线程复用: 线程池中的线程可以被复用,从而减少线程创建的开销。
-
队列管理: 当线程池达到最大线程数时,新任务会被放入队列中等待执行。
-
拒绝策略: 当线程池已满并且队列也已满时,可以配置拒绝策略来处理新任务的方式。
RejectedExecutionHandler 是 Java 线程池的一个重要接口,用于定义当线程池已满并且无法接受新任务时,如何处理被拒绝的任务。当线程池的队列和线程都已满,新任务就会被拒绝执行,这时就会使用 RejectedExecutionHandler 来处理这些被拒绝的任务。
在 Java 中,有几种内置的 RejectedExecutionHandler 实现可供选择,每种实现都有不同的拒绝策略:
AbortPolicy(默认策略): 这是默认的拒绝策略,它会抛出一个 RejectedExecutionException 异常,表示任务被拒绝执行。
CallerRunsPolicy: 当线程池已满时,将任务返回给提交任务的调用者(Caller)。这意味着提交任务的线程会尝试执行被拒绝的任务。
DiscardPolicy: 这个策略会默默地丢弃被拒绝的任务,不会产生任何异常。
DiscardOldestPolicy: 这个策略会丢弃队列中最老的任务,然后尝试将新任务添加到队列中。除了这些内置的策略,你还可以实现自定义的 RejectedExecutionHandler 接口,以定义特定于你应用程序需求的拒绝策略。你可以根据业务需求来决定拒绝策略,比如记录日志、通知管理员、重试等。
@Configuration
public class ThreadPoolConfig {@Beanprivate ThreadPoolTaskExecutor execThreadPool() {ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();pool.setCorePoolSize(50); // 核心线程数pool.setMaxPoolSize(10000); // 最大线程数pool.setQueueCapacity(0); // 等待队列sizepool.setKeepAliveSeconds(60); // 线程最大空闲存活时间pool.setWaitForTasksToCompleteOnShutdown(true);pool.setAwaitTerminationSeconds(60); // 程序shutdown时最多等60秒钟让现存任务结束pool.setRejectedExecutionHandler(new CallerRunsPolicy()); // 拒绝策略return pool;}
}
通过以上方案,我们可以提高消费侧的TPS,同时杜绝重复上报的现象,极大提高数据准确性和用户体验。
相关文章:
Kafka系列之如何提高消费者消费速度
前言 在实际开发过程中,如果使用Kafka处理超大数据量(千万级、亿级)的场景,Kafka消费者的消费速度可能决定系统性能瓶颈。 实现方案 为了提高消费者的消费速度,我们可以采取以下措施: 将主题的分区数量增大,如 20&…...
mac安装Whisper
Whisper 官方git https://github.com/openai/whisper?tabreadme-ov-file 基本上参考官方的安装流程 pip3 install -U openai-whisper pip3 install githttps://github.com/openai/whisper.git pip3 install --upgrade --no-deps --force-reinstall githttps://github.com/…...
Linux:进程概述(什么是进程、进程控制块PCB、并发与并行、进程的状态、进程的相关命令)
进程概述 (1)What(什么是进程) 程序:磁盘上的可执行文件,它占用磁盘、是一个静态概念 进程:程序执行之后的状态,占用CPU和内存,是一个动态概念;每一个进程都有一个对应的进程控制块…...
Unity UGUI 之 坐标转换
本文仅作学习笔记与交流,不作任何商业用途 本文包括但不限于unity官方手册,唐老狮,麦扣教程知识,引用会标记,如有不足还请斧正 本文在发布时间选用unity 2022.3.8稳定版本,请注意分别 前置知识:…...
使用 uPlot 在 Vue 中创建交互式图表
本文由ScriptEcho平台提供技术支持 项目地址:传送门 使用 uPlot 在 Vue 中创建交互式图表 应用场景介绍 uPlot 是一个轻量级、高性能的图表库,适用于创建各种交互式图表。它具有丰富的功能,包括可自定义的轴、网格、刻度和交互性。本篇博…...
SpringBoot 项目配置文件注释乱码的问题解决方案
一、问题描述 在项目的配置文件中,我们写了一些注释,如下所示: 但是再次打开注释会变成乱码,如下所示: 那么如何解决呢? 二、解决方案 1. 点击” File→Setting" 2. 搜索“File Encodings”, 将框…...
TTS如何正确读AI缩写、金额和数字
案例:Tell me whats AI(a i), you need pay $186.30, your card Number is 1 2 3, your work Number is 5 6 7 8...
python基础知识点(蓝桥杯python科目个人复习计划75)
第一题:ip补充 题目描述: 小蓝的ip地址为192.168.*.21,其中*是一个数字,请问这个数字最大可能是多少? import os import sys# 请在此输入您的代码 print("255") 第二题:出现最多的字符 题目描…...
小技巧:如何在已知PDF密码情况下去掉PDF的密码保护
第一步,用Edge打开你的pdf,输入密码进去 第二步,点击打印 第三步,选择导出PDF,选择彩印 第四步,选择导出位置,导出成功后打开发现没有密码限制了!...
Java泛型的介绍和基本使用
什么是泛型 泛型就是将类型参数化,比如定义了一个栈,你必须在定义之前声明这个栈中存放的数据的类型,是int也好是double或者其他的引用数据类型也好,定义好了之后这个栈就无法用来存放其他类型的数据。如果这时候我们想要使用这…...
【C++】动态内存管理与模版
目录 1、关键字new: 1、用法: 2、理解: 3、与malloc的相同与不同: 1、相同: 2、不同: 2、模版初阶: 1、函数模版: 1、概念: 2、关键字:template&…...
MongoDB - 组合聚合阶段:$group、$match、$limit、$sort、$skip、$project、$count
文章目录 1. $group2. $group-> $project2.1 $group2.2 $group-> $project2.3 SpringBoot 整合 MongoDB 3. $match-> $group -> $match3.1 $match3.2 $match-> $group3.3 $match-> $group-> $match3.4 SpringBoot 整合 MongoDB 4. $match-> $group->…...
vue element-ui日期控件传参
前端:Vue element-ui <el-form-item label"过期时间" :rules"[ { required: true, message: 请选择过期时间, trigger: blur }]"><el-date-picker v-model"form.expireTime" type"date" format"yyyy-MM-dd&…...
MacOS安装SDKMan管理Java版本
文章目录 1 简介2 安装与卸载2.1 安装2.2 卸载 3 使用3.1 查看其他工具:支持 Ant, Maven 等3.2 查看Java版本3.3 安装Java,加上相关的版本3.4 设置Java版本(全局)3.5 只在当前窗口生效3.6 卸载1 默认环境无法卸载 4 jdk安装的位置5 与IDEA集成参考 1 简介…...
【网络安全的神秘世界】文件包含漏洞
🌝博客主页:泥菩萨 💖专栏:Linux探索之旅 | 网络安全的神秘世界 | 专接本 | 每天学会一个渗透测试工具 一、概述 文件包含:重复使用的函数写在文件里,需要使用某个函数时直接调用此文件,而无需再…...
并发编程--volatile
1.什么是volatile volatile是 轻 量 级 的 synchronized,它在多 处 理器开 发 中保 证 了共享 变 量的 “ 可 见 性 ” 。可 见 性的意思是当一个 线 程 修改一个共享变 量 时 ,另外一个 线 程能 读 到 这 个修改的 值 。如果 volatile 变 量修 饰 符使用…...
记录unraid docker更新的域名
背景:级联 一、安装内容 unraid更新docker,之前一直失败,修改网络后可以进行安装。 二、查看域名 查看域名,发现是走github的,怪不得有一些docker无法正常更新 三、解决方法 更改代理,这里为unraid的…...
SpringCloud+Vue3多对多,多表联查
♥️作者:小宋1021 🤵♂️个人主页:小宋1021主页 ♥️坚持分析平时学习到的项目以及学习到的软件开发知识,和大家一起努力呀!!! 🎈🎈加油! 加油!…...
麒麟系统信创改造
麒麟系统信创改造 一、查看操作系统架构下载相应的依赖,压缩包1、查看Linux系统架构、CPU(1)uname -m(2)lscpu(3)cat /proc/cpuinfo(4)arch(5)getconf LONG_BIT(6)dmidecode2、根据Linux系统架构、CPU的差异进行下载相关依赖,看第二项二、以下是根据本系统的aarc…...
【Android】ListView和RecyclerView知识总结
文章目录 ListView步骤适配器AdpterArrayAdapterSimpleAdapterBaseAdpter效率问题 RecyclerView具体实现不同布局形式的设置横向滚动瀑布流网格 点击事件 ListView ListView 是 Android 中的一种视图组件,用于显示可滚动的垂直列表。每个列表项都是一个视图对象&…...
深入浅出Asp.Net Core MVC应用开发系列-AspNetCore中的日志记录
ASP.NET Core 是一个跨平台的开源框架,用于在 Windows、macOS 或 Linux 上生成基于云的新式 Web 应用。 ASP.NET Core 中的日志记录 .NET 通过 ILogger API 支持高性能结构化日志记录,以帮助监视应用程序行为和诊断问题。 可以通过配置不同的记录提供程…...
大话软工笔记—需求分析概述
需求分析,就是要对需求调研收集到的资料信息逐个地进行拆分、研究,从大量的不确定“需求”中确定出哪些需求最终要转换为确定的“功能需求”。 需求分析的作用非常重要,后续设计的依据主要来自于需求分析的成果,包括: 项目的目的…...
树莓派超全系列教程文档--(61)树莓派摄像头高级使用方法
树莓派摄像头高级使用方法 配置通过调谐文件来调整相机行为 使用多个摄像头安装 libcam 和 rpicam-apps依赖关系开发包 文章来源: http://raspberry.dns8844.cn/documentation 原文网址 配置 大多数用例自动工作,无需更改相机配置。但是,一…...
基于距离变化能量开销动态调整的WSN低功耗拓扑控制开销算法matlab仿真
目录 1.程序功能描述 2.测试软件版本以及运行结果展示 3.核心程序 4.算法仿真参数 5.算法理论概述 6.参考文献 7.完整程序 1.程序功能描述 通过动态调整节点通信的能量开销,平衡网络负载,延长WSN生命周期。具体通过建立基于距离的能量消耗模型&am…...
【位运算】消失的两个数字(hard)
消失的两个数字(hard) 题⽬描述:解法(位运算):Java 算法代码:更简便代码 题⽬链接:⾯试题 17.19. 消失的两个数字 题⽬描述: 给定⼀个数组,包含从 1 到 N 所有…...
基于Flask实现的医疗保险欺诈识别监测模型
基于Flask实现的医疗保险欺诈识别监测模型 项目截图 项目简介 社会医疗保险是国家通过立法形式强制实施,由雇主和个人按一定比例缴纳保险费,建立社会医疗保险基金,支付雇员医疗费用的一种医疗保险制度, 它是促进社会文明和进步的…...
Unsafe Fileupload篇补充-木马的详细教程与木马分享(中国蚁剑方式)
在之前的皮卡丘靶场第九期Unsafe Fileupload篇中我们学习了木马的原理并且学了一个简单的木马文件 本期内容是为了更好的为大家解释木马(服务器方面的)的原理,连接,以及各种木马及连接工具的分享 文件木马:https://w…...
管理学院权限管理系统开发总结
文章目录 🎓 管理学院权限管理系统开发总结 - 现代化Web应用实践之路📝 项目概述🏗️ 技术架构设计后端技术栈前端技术栈 💡 核心功能特性1. 用户管理模块2. 权限管理系统3. 统计报表功能4. 用户体验优化 🗄️ 数据库设…...
CVE-2020-17519源码分析与漏洞复现(Flink 任意文件读取)
漏洞概览 漏洞名称:Apache Flink REST API 任意文件读取漏洞CVE编号:CVE-2020-17519CVSS评分:7.5影响版本:Apache Flink 1.11.0、1.11.1、1.11.2修复版本:≥ 1.11.3 或 ≥ 1.12.0漏洞类型:路径遍历&#x…...
Python Ovito统计金刚石结构数量
大家好,我是小马老师。 本文介绍python ovito方法统计金刚石结构的方法。 Ovito Identify diamond structure命令可以识别和统计金刚石结构,但是无法直接输出结构的变化情况。 本文使用python调用ovito包的方法,可以持续统计各步的金刚石结构,具体代码如下: from ovito…...
