Kafa分区策略实现
引言
Kafka 的分区策略决定了生产者发送的消息会被分配到哪个分区中,合理的分区策略有助于实现负载均衡、提高消息处理效率以及满足特定的业务需求。
轮询策略(默认)
- 轮询策略是 Kafka 默认的分区策略(当消息没有指定键时)。生产者会按照顺序依次将消息发送到各个分区中,确保每个分区都能均匀地接收到消息,从而实现负载均衡。简单高效,能使各个分区的消息量相对均衡,充分利用每个分区的存储和处理能力。
-
import org.apache.kafka.clients.producer.*; import java.util.Properties;public class RoundRobinProducer {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");Producer<String, String> producer = new KafkaProducer<>(props);for (int i = 0; i < 10; i++) {ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "message-" + i);producer.send(record);}producer.close();} }随机策略
- 随机策略会随机地将消息分配到一个分区中。这种策略在某些情况下可以实现一定程度的负载均衡,但由于是随机分配,可能会导致分区之间的消息分布不够均匀。可以通过自定义分区器来实现随机策略。
-
import org.apache.kafka.clients.producer.*; import java.util.List; import java.util.Map; import java.util.Random;public class RandomPartitioner implements Partitioner {private final Random random = new Random();@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);return random.nextInt(partitions.size());}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {} }// 使用随机分区器的生产者示例 public class RandomProducer {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("partitioner.class", "RandomPartitioner");Producer<String, String> producer = new KafkaProducer<>(props);for (int i = 0; i < 10; i++) {ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "message-" + i);producer.send(record);}producer.close();} }按键哈希策略
- 当消息指定了键时,Kafka 会根据键的哈希值将消息分配到特定的分区中。相同键的消息会被分配到同一个分区,这有助于保证具有相同业务逻辑的消息顺序性。可以保证消息的局部有序性,例如在处理用户相关的消息时,将同一个用户的消息发送到同一个分区,方便后续的处理和分析。
-
import org.apache.kafka.clients.producer.*; import java.util.Properties;public class KeyBasedProducer {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");Producer<String, String> producer = new KafkaProducer<>(props);for (int i = 0; i < 10; i++) {ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "user-" + (i % 2), "message-" + i);producer.send(record);}producer.close();} }自定义分区策略(实现接口)
-
当上述默认策略无法满足业务需求时,可以自定义分区策略。通过实现
org.apache.kafka.clients.producer.Partitioner接口,重写partition方法来实现自定义的分区逻辑。例如,根据消息的某些特定字段(如时间、地理位置等)来进行分区,以满足特定的业务需求。 -
import org.apache.kafka.clients.producer.*; import java.util.List; import java.util.Map;public class CustomPartitioner implements Partitioner {@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);// 自定义分区逻辑,这里简单示例根据消息值的长度分区String message = (String) value;return message.length() % partitions.size();}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {} }// 使用自定义分区器的生产者示例 public class CustomProducer {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("partitioner.class", "CustomPartitioner");Producer<String, String> producer = new KafkaProducer<>(props);for (int i = 0; i < 10; i++) {ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "message-" + i);producer.send(record);}producer.close();} }
相关文章:
Kafa分区策略实现
引言 Kafka 的分区策略决定了生产者发送的消息会被分配到哪个分区中,合理的分区策略有助于实现负载均衡、提高消息处理效率以及满足特定的业务需求。 轮询策略(默认) 轮询策略是 Kafka 默认的分区策略(当消息没有指定键时&…...
Pyside/Pyqt中QWebEngineView和QWebEnginePage的区别
在 PySide/Qt 的 WebEngine 模块中,QWebEngineView 和 QWebEnginePage 是两个紧密相关但职责不同的类。以下是它们的核心区别和关系: 1. 职责区分 类名核心职责模块归属QWebEngineView作为可视化的窗口部件(Widget),负…...
Kafka的内部通信协议
引言 kafka内部用到的常见协议和优缺点可以看看原文 Kafka用到的协议 本文奖详细探究kafka核心通信协议和高性能的关键 网络层通信的实现 基于 Java NIO:Kafka 的网络通信层主要基于 Java NIO 来实现,这使得它能够高效地处理大量的连接和数据传输。…...
强大到工业层面的软件
电脑数据删不干净,简直是一种让人抓狂的折磨!明明已经把文件扔进了回收站,清空了,可那些残留的数据就像牛皮癣一样,怎么也除不掉。这种烦恼简直无处不在,让人从头到脚都感到无比烦躁。 首先,心…...
数据分析和AI丨应对AI实施挑战,工程领域AI应用的五大方法
工程领域的人工智能 (AI) 已经开始发挥价值,低代码和无代码工具正在使曾经仅属于专业数据科学家的 AI 能力变得大众化。 然而,并非工程领域的每个人都能从中受益,使用新的便捷的 AI 工具提高工作效率并不难,…...
54. UDP协议
UDP协议 UDP(User Datagram Protocol,用户数据报协议)是一个无连接的传输层协议,它提供简单的、不可靠的信息传送服务。与TCP(传输控制协议)不同,UDP不提供数据包的排序、错误检查(仅…...
AJAX笔记入门篇
黑马程序员视频地址: 黑马程序员前端AJAX入门到实战全套教程https://www.bilibili.com/video/BV1MN411y7pw?vd_source0a2d366696f87e241adc64419bf12cab&spm_id_from333.788.videopod.episodes&p2https://www.bilibili.com/video/BV1MN411y7pw?vd_source…...
深入解析Java集合框架:春招面试要点
在上一篇文章中,我们深入探讨了Java核心基础,这是学习Java的基石。而在实际的Java开发中,集合框架的使用频率极高,它为我们提供了丰富的数据结构和算法实现,极大地提高了开发效率。对于春招面试来说,集合框…...
【Elasticsearch】Elasticsearch的查询
Elasticsearch的查询 DSL查询基础语句叶子查询全文检索查询matchmulti_match 精确查询termrange 复合查询算分函数查询bool查询 排序分页基础分页深度分页 高亮高亮原理实现高亮 RestClient查询基础查询叶子查询复合查询排序和分页高亮 数据聚合DSL实现聚合Bucket聚合带条件聚合…...
STM32 PWM驱动直流电机
接线图: 代码配置: 根据驱动舵机的代码来写,与舵机不同的是,这次的引脚接到了PA2上,所以需要改一下引脚以及改为OC3通道。 另外还需在配置两个GPIO引脚,来控制电机的旋转方向,这里连接到了PA4与…...
系统思考—心智模式
“我们的大脑对连贯性的渴望远胜于对准确性的追求。”—诺贝尔经济学得主丹尼尔卡尼曼 在面对复杂的决策时,我们往往更倾向于寻找那些能够迅速串联起来的信息,而非深入挖掘每一个细节的真实性。这种倾向在日常生活中或许能帮助我们迅速作出决策…...
JavaScript_02 表单
表单常用演示: 1.图片 结果失真了... 2.切换图片 切换结果 3.表单:...
【Qt】06-对话框
对话框 前言一、模态和非模态对话框1.1 概念1.2 模态对话框1.2.1 代码QAction类 1.2.2 模态对话框运行分析 1.3 非模态对话框1.3.1 代码局部变量和成员变量setAttribute 类 1.3.2 现象解释 二、标准对话框2.1 提示对话框 QMessageBox2.1.1 现象及解释 2.2 问题对话框2.2.1 现象…...
AI学习指南Ollama篇-使用Ollama构建自己的私有化知识库
一、引言 (一)背景介绍 随着企业对数据隐私和效率的重视,私有化知识库的需求日益增长。私有化知识库不仅可以保护企业数据的安全性,还能提供高效的知识管理和问答系统,提升企业内部的工作效率和创新能力。 (二)Ollama和AnythingLLM的结合 Ollama和AnythingLLM的结合…...
2.策略模式(Strategy)
定义 定义一系列算法,把它们一个个封装起来,并且使他们可互相替换(变化)。该模式使算法可独立于使用它的客户程序(稳定)而变化(拓展,子类化)。 动机(Motiva…...
Python里的小整数问题挺有意思的
简单来说,Python为了优化性能,会把一些常用的整数(通常是-5到256)提前创建好,放到一个“缓存池”里。这样,当你用到这些小整数时,Python就不用每次都重新创建对象了,直接从缓存池里拿…...
开源智慧园区管理系统对比五款主流产品探索智能运营新模式
内容概要 在这个数字化迅速发展的时代,园区管理也迎来了全新的机遇和挑战。众所周知,开源智慧园区管理系统作为一种创新解决方案,正逐步打破传统管理的局限性。它的开放性不仅使得系统可以根据具体需求进行灵活调整,也为用户提供…...
正则表达式入门
入门 1、提取文章中所有的英文单词 //1.先创建一个Pattern对象,模式对象,可以理解成就是一个正则表达式对象 Pattern pattern Pattern.compile("[a-zA-Z]"); //2.创建一个匹配器对象 //理解:就是 matcher匹配器按照p…...
hive:数据导入,数据导出,加载数据到Hive,复制表结构
hive不建议用insert,因为Hive是建立在Hadoop之上的数据仓库工具,主要用于批处理和大数据分析,而不是为OLTP(在线事务处理)操作设计的。INSERT操作会非常慢 数据导入 命令行界面:建一个文件 查询数据>>复制>>粘贴到新…...
【某大厂一面】HashSet底层怎么实现的
HashSet 是 Java 集合框架中的一个非常常用的集合类,它实现了 Set 接口,并且底层通常是通过 哈希表(HashMap)来实现的。要理解 HashSet 的底层实现,我们需要从哈希表的工作原理开始讲起。下面是对 HashSet 底层实现的详…...
idea大量爆红问题解决
问题描述 在学习和工作中,idea是程序员不可缺少的一个工具,但是突然在有些时候就会出现大量爆红的问题,发现无法跳转,无论是关机重启或者是替换root都无法解决 就是如上所展示的问题,但是程序依然可以启动。 问题解决…...
React 第五十五节 Router 中 useAsyncError的使用详解
前言 useAsyncError 是 React Router v6.4 引入的一个钩子,用于处理异步操作(如数据加载)中的错误。下面我将详细解释其用途并提供代码示例。 一、useAsyncError 用途 处理异步错误:捕获在 loader 或 action 中发生的异步错误替…...
python/java环境配置
环境变量放一起 python: 1.首先下载Python Python下载地址:Download Python | Python.org downloads ---windows -- 64 2.安装Python 下面两个,然后自定义,全选 可以把前4个选上 3.环境配置 1)搜高级系统设置 2…...
dedecms 织梦自定义表单留言增加ajax验证码功能
增加ajax功能模块,用户不点击提交按钮,只要输入框失去焦点,就会提前提示验证码是否正确。 一,模板上增加验证码 <input name"vdcode"id"vdcode" placeholder"请输入验证码" type"text&quo…...
微服务商城-商品微服务
数据表 CREATE TABLE product (id bigint(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 商品id,cateid smallint(6) UNSIGNED NOT NULL DEFAULT 0 COMMENT 类别Id,name varchar(100) NOT NULL DEFAULT COMMENT 商品名称,subtitle varchar(200) NOT NULL DEFAULT COMMENT 商…...
JS手写代码篇----使用Promise封装AJAX请求
15、使用Promise封装AJAX请求 promise就有reject和resolve了,就不必写成功和失败的回调函数了 const BASEURL ./手写ajax/test.jsonfunction promiseAjax() {return new Promise((resolve, reject) > {const xhr new XMLHttpRequest();xhr.open("get&quo…...
【Linux系统】Linux环境变量:系统配置的隐形指挥官
。# Linux系列 文章目录 前言一、环境变量的概念二、常见的环境变量三、环境变量特点及其相关指令3.1 环境变量的全局性3.2、环境变量的生命周期 四、环境变量的组织方式五、C语言对环境变量的操作5.1 设置环境变量:setenv5.2 删除环境变量:unsetenv5.3 遍历所有环境…...
vue3 daterange正则踩坑
<el-form-item label"空置时间" prop"vacantTime"> <el-date-picker v-model"form.vacantTime" type"daterange" start-placeholder"开始日期" end-placeholder"结束日期" clearable :editable"fal…...
API网关Kong的鉴权与限流:高并发场景下的核心实践
🔥「炎码工坊」技术弹药已装填! 点击关注 → 解锁工业级干货【工具实测|项目避坑|源码燃烧指南】 引言 在微服务架构中,API网关承担着流量调度、安全防护和协议转换的核心职责。作为云原生时代的代表性网关,Kong凭借其插件化架构…...
小木的算法日记-多叉树的递归/层序遍历
🌲 从二叉树到森林:一文彻底搞懂多叉树遍历的艺术 🚀 引言 你好,未来的算法大神! 在数据结构的世界里,“树”无疑是最核心、最迷人的概念之一。我们中的大多数人都是从 二叉树 开始入门的,它…...
