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

Kafka生产者的粘性分区算法

分区算法分类

kafka在生产者投递消息时,会根据是否有key采取不用策略来获取分区。

存在key时会根据key计算一个hash值,然后采用hash%分区数的方式获取对应的分区。

而不存在key时采用随机算法选取分区,然后将所有的消息封装到这个batch上直到达到限定数量,然后才发送出去。

如下图,6条消息采用key可能分三次发送到三个不同的分区,需要3次网络请求。如果没有key将封住成一个批次发送。这样一次网路请求就可以发送多条消息,大大提高了效率。

源码分析

producer根据keyBytes是否有值采用不同的分区策略。有key的计算hash % numPartitions得到分区。

 public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster,int numPartitions) {if (keyBytes == null) {return stickyPartitionCache.partition(topic, cluster);}// hash the keyBytes to choose a partitionreturn Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;}

 并且kafka在这里做了缓存,如果第一次获取到了粘性分区后面会缓存起来。 

 public int partition(String topic, Cluster cluster) {Integer part = indexCache.get(topic);if (part == null) {return nextPartition(topic, cluster, -1);}return part;}

没有key的采用stickyPartitionCache的策略,这里是分区算法的主要代码。获取所有的availablePartitions,然后如果availablePartitions大于1,获取一个随机数random,然后通过random % availablePartitions.size()的方式获取分区。

      List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);if (availablePartitions.size() < 1) {Integer random = Utils.toPositive(ThreadLocalRandom.current().nextInt());newPart = random % partitions.size();} else if (availablePartitions.size() == 1) {newPart = availablePartitions.get(0).partition();} else {while (newPart == null || newPart.equals(oldPart)) {int random = Utils.toPositive(ThreadLocalRandom.current().nextInt());newPart = availablePartitions.get(random % availablePartitions.size()).partition();}}

abortForNewBatch表示需要发送到新的批次,然后调用onNewBatch获取新的分区。

      if (result.abortForNewBatch) {int prevPartition = partition;partitioner.onNewBatch(record.topic(), cluster, prevPartition);partition = partition(record, serializedKey, serializedValue, cluster);tp = new TopicPartition(record.topic(), partition);...public void onNewBatch(String topic, Cluster cluster, int prevPartition) {stickyPartitionCache.nextPartition(topic, cluster, prevPartition);}

在下一个批次发送时会检测是否和上一个分区相同,如果相同将会缓存一个新的分区。

        // Check that the current sticky partition for the topic is either not set or that the partition that // triggered the new batch matches the sticky partition that needs to be changed.if (oldPart == null || oldPart == prevPartition) {

总结

为了提升kafka发送消息的速率,在对消息顺序没有特殊的要求情况下,应该尽量避免设置消息的key,这样可以提交发送消息的吞吐量。

相关文章:

Kafka生产者的粘性分区算法

分区算法分类 kafka在生产者投递消息时&#xff0c;会根据是否有key采取不用策略来获取分区。 存在key时会根据key计算一个hash值&#xff0c;然后采用hash%分区数的方式获取对应的分区。 而不存在key时采用随机算法选取分区&#xff0c;然后将所有的消息封装到这个batch上直…...

java基础篇

1.基础篇注释注释是在程序指定位置添加的说明性信息注释不参与程序运行&#xff0c;仅起到说明作用单行注释 格式&#xff1a;// 注释信息多行注释 格式&#xff1a;/* 注释信息 */关键字关键字&#xff1a;就是被Java语言赋予了特定含义的单词java中共有53个关键字1.全部有小写…...

Java与Winform进行AES加解密数据传输的工具类与对应关系和示例

场景 AndroidJava中使用Aes对称加密的工具类与使用&#xff1a; AndroidJava中使用Aes对称加密的工具类与使用_霸道流氓气质的博客-CSDN博客 上面讲的Java与安卓进行数据传输时使用AES加解密的示例工具类。 如果Java需要与其他第三方平台比如Winform程序进行数据传递时也需…...

OpenAI模型的API调用与使用-测试(2)

OpenAI模型的API调用与使用-测试&#xff08;2&#xff09;1. 参考Quick start搭建一个demo1.1 安装openai包1.2 demo测试11.3 demo测试2参考资料1. 参考Quick start搭建一个demo 1.1 安装openai包 注意关掉科学上网工具&#xff0c;下载openai包 pip install openai安装好后…...

【LeetCode】剑指 Offer 22. 链表中倒数第k个节点 p136 -- Java Version

题目链接&#xff1a;https://leetcode.cn/problems/lian-biao-zhong-dao-shu-di-kge-jie-dian-lcof/ 1. 题目介绍&#xff08;22. 链表中倒数第k个节点&#xff09; 输入一个链表&#xff0c;输出该链表中倒数第k个节点。为了符合大多数人的习惯&#xff0c;本题从1开始计数&…...

经典卷积模型回顾7-轻量化模型MobileNet实现图像分类(matlab)

MobileNet是一种轻量级卷积神经网络&#xff0c;适用于较小的设备和低功耗环境。在MATLAB中&#xff0c;可以使用Deep Learning Toolbox进行MobileNet的图像分类训练。 使用预先训练好的MobileNet模型对自定义数据集进行微调训练&#xff1a; matlab % 导入数据集 imds im…...

程序员压力大?用 PyQt 做一个美*女GIF设置桌面,每天都有好心情

嗨害大家好鸭&#xff01;我是小熊猫~ 要说程序员工作的最大压力不是来自于工作本身&#xff0c; 而是来自于需要不断学习才能更好地完成工作&#xff0c; 因为程序员工作中面对的编程语言是在不断更新的&#xff0c; 同时还要学习熟悉其他语言来提升竞争力… 好了&#xff0c…...

Shell命令——sed命令

以下内容整理于《linux命令行与shell脚本编程大全【第三版】》一书。 一、简介sed编辑器 1、sed编辑器的本质 sed是stream editor的缩写&#xff0c;中文意思是“流编辑器”。 sed编辑器是一个命令行编辑器&#xff0c;也就是可以在命令行上完成数据的处理&#xff08;替换、…...

C语言练习 | 初学者经典练习汇(2)

目录 1、编写一个程序从1到100中&#xff0c;所有出现9的个数 2、分数求和 3、10个整形数字中选出最大值 4、打印9*9的乘法口诀 5、字符串逆序 6、计算一个数的每位之和(递归实现) 7、递归实现n的K次方 8、写个冒泡排序&#xff0c;把一个整形数组变成升序。 9、二进制…...

git分支

分支什么是分支在版本控制过程中&#xff0c;同时推进多个任务&#xff0c;为每个任务&#xff0c;我们就可以创建每个任务的单独分支。使用分支意味着程序员可以把自己的工作从开发主线上分离开来&#xff0c;开发自己分支的时候&#xff0c;不会影响主线分支的运行。对于初学…...

Java每天15道面试题 | redisII

1、什么是 Redis&#xff1f;简述它的优缺点&#xff1f; Redis 本质上是一个 Key-Value 类型的内存数据库&#xff0c;很像 memcached&#xff0c;整个数据库统统加载在内存当中进行操作&#xff0c;定期通过异步操作把数据库数据 flush 到硬盘上进行保存。因为是纯内存操作&a…...

浏览器渲染原理

阶段 - Parse 1、解析HTML&#xff0c;浏览器将从服务器获取到的HTML文件之后&#xff0c;会产生一个渲染任务&#xff0c;交给消息队列&#xff08;EventLoop/MessageLoop&#xff09;。 2、在事件循环机制的作用下&#xff0c;会将渲染任务交给主线程 3、主线程在获取到渲染…...

华为OD机试题 - 查找单入口空闲区域(JavaScript)| 含思路

华为OD机试题 最近更新的博客使用说明本篇题解:查找单入口空闲区域题目输入输出示例一输入输出说明示例二输入输出说明示例三输入输出说明示例四输入输出说明Code解题思路华为OD其它语言版本<...

制造型企业想要做好数字化改造,要注意以下几点!

很多企业在“工业4.0、智能制造、互联网”等概念满天飞的环境下迷失了方向&#xff0c;不知该如何下手&#xff0c;盲目跟风&#xff0c;看别人投自动化&#xff0c;自己也跟着投&#xff0c;看别人上信息化&#xff0c;自己也跟着上。 其实&#xff0c;智能制造也好&#xff…...

【蓝桥杯集训·每日一题】AcWing 1488. 最短距离

文章目录一、题目1、原题链接2、题目描述二、解题报告1、思路分析2、时间复杂度3、代码详解三、知识风暴Dijkstra算法一、题目 1、原题链接 1488. 最短距离 2、题目描述 有 N 个村庄&#xff0c;编号 1 到 N。 村庄之间有 M 条无向道路&#xff0c;第 i 条道路连接村庄 ai 和村…...

比亚迪:全球最大电动汽车制造商的坎坷成长之路

来源&#xff1a;猛兽财经 作者&#xff1a;猛兽财经 特斯拉&#xff08;TSLA&#xff09;首席执行官埃隆马斯克表示&#xff0c;特斯拉最接近的竞争对手可能是一家中国电动汽车公司。猛兽财经认为&#xff0c;沃伦•巴菲特支持的比亚迪&#xff08;0211&#xff09;可能是马斯…...

Java开发 - Quartz初体验

前言 在上一篇博客中&#xff0c;我们对单点登录有了初步了解&#xff0c;这也让我们独立做系统有了最基础的保障。但在业务开发中&#xff0c;总是会出现一些定期处理的任务&#xff0c;我们首先想到的是Timer&#xff0c;但由于其调度功能单一&#xff0c;我们实际并不会用它…...

无头盔开发vr XR Device Simulator操作(更新)

1.摄像机&#xff08;未开启TY键&#xff09; 平移 上下左右&#xff1a;右键鼠标&#xff0c;移哪去哪 前后&#xff1a;右键快速滚动鼠标滚轮 旋转 XOY平面旋转&#xff1a;右键按住鼠标滚轮滚动鼠标滚轮 XOZ\YOZ平面旋转&#xff1a;右键按住鼠标滚轮移动鼠标 2.左手右手&am…...

《C++代码分析》第二回:函数重载const char* ,char*,const char[],char[]汇编代码上的区别

一、前言 C相比C是支持函数重载的&#xff0c;现在我们详细探讨一下C函数重载与类方法承载。 二、案例代码 我们编译如下代码&#xff0c;同样的我们关闭代码优化&#xff0c;删除符号链接文&#xff08;.pdb&#xff09; #include "windows.h" #include "w…...

【学习笔记】深入理解JVM之垃圾回收机制

【学习笔记】深入理解JVM之垃圾回收机制 更多文章首发地址&#xff1a;地址 参考&#xff1a; 《深入理解JAVA虚拟机》第三版 第三章尚硅谷 第134 - 203 集参考文章&#xff1a;https://blog.csdn.net/qq_48435252/article/details/123697193 1、概念 &#x1f33b; 首先我们…...

Linux应用开发之网络套接字编程(实例篇)

服务端与客户端单连接 服务端代码 #include <sys/socket.h> #include <sys/types.h> #include <netinet/in.h> #include <stdio.h> #include <stdlib.h> #include <string.h> #include <arpa/inet.h> #include <pthread.h> …...

Oracle查询表空间大小

1 查询数据库中所有的表空间以及表空间所占空间的大小 SELECTtablespace_name,sum( bytes ) / 1024 / 1024 FROMdba_data_files GROUP BYtablespace_name; 2 Oracle查询表空间大小及每个表所占空间的大小 SELECTtablespace_name,file_id,file_name,round( bytes / ( 1024 …...

【大模型RAG】Docker 一键部署 Milvus 完整攻略

本文概要 Milvus 2.5 Stand-alone 版可通过 Docker 在几分钟内完成安装&#xff1b;只需暴露 19530&#xff08;gRPC&#xff09;与 9091&#xff08;HTTP/WebUI&#xff09;两个端口&#xff0c;即可让本地电脑通过 PyMilvus 或浏览器访问远程 Linux 服务器上的 Milvus。下面…...

连锁超市冷库节能解决方案:如何实现超市降本增效

在连锁超市冷库运营中&#xff0c;高能耗、设备损耗快、人工管理低效等问题长期困扰企业。御控冷库节能解决方案通过智能控制化霜、按需化霜、实时监控、故障诊断、自动预警、远程控制开关六大核心技术&#xff0c;实现年省电费15%-60%&#xff0c;且不改动原有装备、安装快捷、…...

什么是库存周转?如何用进销存系统提高库存周转率?

你可能听说过这样一句话&#xff1a; “利润不是赚出来的&#xff0c;是管出来的。” 尤其是在制造业、批发零售、电商这类“货堆成山”的行业&#xff0c;很多企业看着销售不错&#xff0c;账上却没钱、利润也不见了&#xff0c;一翻库存才发现&#xff1a; 一堆卖不动的旧货…...

OkHttp 中实现断点续传 demo

在 OkHttp 中实现断点续传主要通过以下步骤完成&#xff0c;核心是利用 HTTP 协议的 Range 请求头指定下载范围&#xff1a; 实现原理 Range 请求头&#xff1a;向服务器请求文件的特定字节范围&#xff08;如 Range: bytes1024-&#xff09; 本地文件记录&#xff1a;保存已…...

Cloudflare 从 Nginx 到 Pingora:性能、效率与安全的全面升级

在互联网的快速发展中&#xff0c;高性能、高效率和高安全性的网络服务成为了各大互联网基础设施提供商的核心追求。Cloudflare 作为全球领先的互联网安全和基础设施公司&#xff0c;近期做出了一个重大技术决策&#xff1a;弃用长期使用的 Nginx&#xff0c;转而采用其内部开发…...

Spring Boot面试题精选汇总

&#x1f91f;致敬读者 &#x1f7e9;感谢阅读&#x1f7e6;笑口常开&#x1f7ea;生日快乐⬛早点睡觉 &#x1f4d8;博主相关 &#x1f7e7;博主信息&#x1f7e8;博客首页&#x1f7eb;专栏推荐&#x1f7e5;活动信息 文章目录 Spring Boot面试题精选汇总⚙️ **一、核心概…...

tree 树组件大数据卡顿问题优化

问题背景 项目中有用到树组件用来做文件目录&#xff0c;但是由于这个树组件的节点越来越多&#xff0c;导致页面在滚动这个树组件的时候浏览器就很容易卡死。这种问题基本上都是因为dom节点太多&#xff0c;导致的浏览器卡顿&#xff0c;这里很明显就需要用到虚拟列表的技术&…...

力扣-35.搜索插入位置

题目描述 给定一个排序数组和一个目标值&#xff0c;在数组中找到目标值&#xff0c;并返回其索引。如果目标值不存在于数组中&#xff0c;返回它将会被按顺序插入的位置。 请必须使用时间复杂度为 O(log n) 的算法。 class Solution {public int searchInsert(int[] nums, …...