kafka发送大消息

1 kafka消息压缩

kafka关于消息压缩的定义（来源于官网）：

        此为 Kafka 中端到端的块压缩功能。如果启用，数据将由 producer 压缩，以压缩格式写入服务器，并由 consumer 解压缩。压缩将提高 consumer 的吞吐量，但需付出一定的解压成本。

        压缩就是用时间换空间，其基本理念是基于重复，将重复的片段编码为字典，字典的 key 为重复片段，value 为更短的代码，比如序列号，然后将原始内容中的片段用代码表示，达到缩短内容的效果，压缩后的内容则由字典和代码序列两部分组成。解压时根据字典和代码序列可无损地还原为原始内容。通常来讲，重复越多，压缩效果越好。比如 JSON 是 Kafka 消息中常用的序列化格式，单条消息内可能并没有多少重复片段，但如果是批量消息，则会有大量重复的字段名，批量中消息越多，则重复越多，这也是为什么 Kafka 更偏向块压缩，而不是单条消息压缩。

2 kafka的消息压缩类型对比

        目前 Kafka 共支持四种主要的压缩类型：Gzip、Snappy、Lz4 和 Zstd。关于这几种压缩的特性。

压缩类型压缩率CPU 使用率压缩速度带宽使用率
GzipHighestHighestSlowestLowest
SnappyMediumModerateModerateMedium
Lz4LowLowestFastestHighest
ZstdMediumModerateModerateMedium

        从上表可知，Snappy 在 CPU 使用率、压缩比、压缩速度和网络带宽使用率之间实现良好的平衡，我们最终也是采用的该类型进行压缩试点。这里值得一提的是，Zstd 是 Facebook 于 2016 年开源的新压缩算法，压缩率和压缩性能都不错，具有与 Snappy（Google 杰作）相似的特性，直到 Kafka 的 2.1.0 版本才引入支持。

3 何时需要压缩

        压缩是需要额外的 CPU 代价的，并且会带来一定的消息分发延迟，因而在压缩前要慎重考虑是否有必要。

  • 压缩带来的磁盘空间和带宽节省远大于额外的 CPU 代价，这样的压缩是值得的。
  • 数据量足够大且具重复性。消息压缩是批量的，低频的数据流可能都无法填满一个批量，会影响压缩比。数据重复性越高，往往压缩效果越好，例如 JSON、XML 等结构化数据；但若数据不具重复性，例如文本都是唯一的 md5 或 UUID 之类，违背了压缩的重复性前提，压缩效果可能不会理想。
  • 系统对消息分发的延迟没有严苛要求，可容忍轻微的延迟增长。

4 如何开启压缩

        Kafka 通过配置属性 compression.type 控制是否压缩。该属性在 producer 端和 broker 端各自都有一份，也就是说，我们可以选择在 producer 或 broker 端开启压缩，对应的应用场景各有不同。目前没有尝试在broker段开启压缩。

4.1 在broker端开启解压缩

        Broker 端的 compression.type 属性默认值为 producer，即直接继承 producer 端所发来消息的压缩方式，无论消息采用何种压缩或者不压缩，broker 都原样存储。、

4.1.1 broker 和 topic 两个级别

        在 broker 端的压缩配置分为两个级别：全局的 broker 级别 和 局部的 topic 级别。顾名思义，如果配置的是 broker 级别，则对于该 Kafka 集群中所有的 topic 都是生效的。但如果 topic 级别配置了自己的压缩类型，则会覆盖 broker 全局的配置，以 topic 自己配置的为准。

broker级别：要配置 broker 级别的压缩类型，可通过 configs 命令修改   compression.type  配置项取值。此处要使修改生效，是否需要重启 broker 取决于 Kafak 的版本，在 1.1.0 之前，任何配置项的改动都需要重启 broker 才生效，而从 1.1.0 版本开始，Kafka 引入了动态 broker 参数，将配置项分为三类：read-only、per-broker 和 cluster-wide，第一类跟原来一样需重启才生效，而后面两类都是动态生效的，只是影响范围不同，关于 Kafka 动态参数，以后单开博文介绍。从 官网 可以看到，compression.type 是属于 cluster-wide 的，如果是 1.1.0 及之后的版本，则无需重启 broker。

topic级别：topic 的配置分为两部分，一部分是 topic 特有的，如 partitions 等，另一部分则是默认采用 broker 配置，但也可以覆盖。如果要定义 topic 级别的压缩，可以在 topic 创建时通过 --config 选项覆盖配置项 compression.type 的取值，命令如下：

sh bin/kafka-topics.sh \
--create \
--topic my-topic \
--replication-factor 1 \
--partitions 1 \
--config compression.type=snappy

也可以通过 configs 命令修改 topic 的 compression.type 取值，命令如下：

bin/kafka-configs.sh \
--entity-type topics \
--entity-name my-topic \
--alter \
--add-config compression.type=snappy

4.2 在 Producer 端压缩

        跟 broker 端一样，producer 端的压缩配置属性依然是 compression.type，只不过默认值和可选值有所不同。默认值为 none，表示不压缩。直接在代码层面更改 producer 的 config。但需要注意的是，改完 config 之后，需要重启 producer 端的应用程序，压缩才会生效。

代码示例如下：

public class KafkaProducerTest {public static void main(String[] args) {String brokerList = "127.0.0.1:9092";Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,"2097245");properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"gzip");KafkaProducer<String, String> producer = new KafkaProducer<>(properties);String topic = "mytestTopic1";int sizeInMb = 2; // 设置字符串大小为2MBint sizeInBytes = sizeInMb * 1024 * 1024; // 转换为字节数StringBuilder largeString = new StringBuilder(sizeInBytes);largeString.append(":");for (int i = 0; i < sizeInBytes; i++) {largeString.append("A"); // 使用大写字母"A"来构建字符串}String msg = largeString.toString();try {for (int i = 0; i < 100; i++) {String msg1 = i+msg;producer.send(new ProducerRecord<>(topic, msg1));Thread.sleep(500);}}catch (Exception e){e.printStackTrace();}}
}

上面示例特意制造了一个大字符串作为消息，测试压缩，需要注意的是，配置压缩的时候同时也需要配置消息的最大值。即：max.request.size。

5 解压缩

可能发生解压的地方依然是两处：consumer 端和 broker 端。

consumer端：consumer 端发生解压的唯一条件就是从 broker 端拉取到的消息是带压缩的。此时，consumer 会根据 recordBatch 中 compressionType 来对消息进行解压。

broker端：broker 端是否发生解压取决于 producer 发过来的批量消息 recordBatch 是否是压缩的：如果 producer 开启了压缩，则会发生解压，否则不会。原因简单说下，在 broker 端持久化消息前，会对消息做各种验证，此时必然会迭代 recordBatch，而在迭代的过程中，会直接采用 recordBatch 上的 compressionType 对消息字节流进行处理，是否解压取决于 compressionType 是否是压缩类型。关于这点，可以在   LogValidator 的 validateMessagesAndAssignOffsets 方法实现中可以看到，在 convertAndAssignOffsetsNonCompressed、assignOffsetsNonCompressed  和 validateMessagesAndAssignOffsetsCompressed 三个不同的分支中，都会看到 records.batches.forEach {...} 的身影，而在后面的源码分析中会发现，在 recordBatch 的迭代器逻辑中，直接采用的 compressionType 的解压逻辑对消息字节流读取的。也就是说，如果 recordBatch 是压缩的 ，只要对其进行了迭代访问，则会自动触发解压逻辑。

通俗一点讲：producer端配置了压缩，consumer自动解压缩。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/578159.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

使用 pytest.ini 文件控制输出 log 日志

使用 pytest.ini 文件控制输出 log 日志

一、前置说明 pytest.ini 文件中可以配置参数来控制 pytest 的运行行为，其存放路径要求与 conftest.py 一样。 项目根目录project_root/ ├── pytest.ini ├── tests/ │ └── test_demo.py以test开头的测试子目录project_root/ ├── tests/ │ ├── pytest.in…
阅读更多...
c# OpenCvSharp透视矫正六步实现透视矫正（八）

c# OpenCvSharp透视矫正六步实现透视矫正（八）

透视矫正,引用文档拍照扫描&#xff0c;相片矫正这块。 读取图像Cv2.ImRead();预处理&#xff08;灰度化&#xff0c;高斯滤波、边缘检测&#xff09;轮廓检测&#xff08;获取到最大轮廓&#xff09;获取最大面积轮廓的四个顶点标识最小矩形坐标透视矫正显示 完整代码 // 1、…
阅读更多...
Linux系统安装及管理

Linux系统安装及管理

目录 一、Linux应用程序基础 1.1应用程序与系统命令的关系 1.2典型应用程序的目录结构 1.3常见的软件包装类型 二、RPM软件包管理 1.RPM是什么&#xff1f; 2.RPM命令的格式 2,1查看已安装的软件包格式 2.2查看未安装的软件包 3.RPM安装包从哪里来&#xff1f; 4.挂…
阅读更多...
web前端项目-七彩夜空烟花【附源码】

web前端项目-七彩夜空烟花【附源码】

web前端项目-七彩动态夜空烟花【附源码】 本项目仅使用了HTML&#xff0c;代码简单&#xff0c;实现效果绚丽&#xff0c;且本项目代码直接运行即可实现&#xff0c;无需图片素材&#xff0c;接下来让我们一起实现一场美丽的烟花秀叭 运行效果&#xff1a;鼠标点击和移动可控制…
阅读更多...
理解SpringMVC的工作流程

理解SpringMVC的工作流程

组件 前置控制器 DispatcherServlet。 映射控制器 HandlerMapping。 处理器 Controller。 模型和视图 ModelAndView。 视图解析器 ViewResolver。 工作流程 spring mvc 先将请求发送给 DispatcherServlet。 DispatcherServlet 查询一个或多个 HandlerMapping&#xff0c;找到…
阅读更多...
vc 用MySQL Connector/C++

vc 用MySQL Connector/C++

1 下载 MySQL :: Download Connector/C 2 vc配置 添加路径 2.1 右击项目 -> 属性 2.2 配置属性-> vc目录 -> 包含目录 -> 添加 D:\mysql-connector-c-8.2.0-winx64\include\jdbc 具体目录在mysql-connector-c的文件夹中搜索 mysql_driver.h文件 然后把这个文件…
阅读更多...
jmeter-set up先登录获取token，再测试

jmeter-set up先登录获取token，再测试

一、何为setup 一种特殊类型的线程组&#xff0c;可用于执行预测试操作&#xff1b;简单来讲就是执行测试线程组前&#xff0c;先执行setup 作用 例如前面&#xff0c;我们说到的&#xff0c;压测之前只用JMeter调用业务接口造数或者通过JDBC操作数据库造数&#xff0c;可以放…
阅读更多...
Spring中BeanFactoryPostProcessors的使用和原理

Spring中BeanFactoryPostProcessors的使用和原理

Spring中的BeanFactoryPostProcessor是在Spring容器实例化Bean之后&#xff0c;初始化之前执行的一个扩展机制。它允许开发者在Bean的实例化和初始化之前对BeanDefinition进行修改和处理&#xff0c;从而对Bean的创建过程进行干预和定制化。 BeanFactoryPostProcessor接口定义…
阅读更多...
信息网络协议基础-接入网技术

信息网络协议基础-接入网技术

文章目录 概述***基于ATM架构虚电路PVC和SVC信元格式为什么信元格式由AAL决定?网络架构传统电信网络:点对点链路PPP协议协议内容消息过程多协议封装功能电话网接入Internet(DSL 数字用户线路)主要接入技术ADSL关键技术DMTDSLAM体系结构PPPOE帧格式过程特点局域网定义参考模型L…
阅读更多...
网络安全法规和模型

网络安全法规和模型

基础 ISO信息安全&#xff1a;为数据处理系统建立和采取技术、管理的安全保护&#xff0c;保护计算机硬件、软件、数据不因偶然的或恶意的原因而受到破坏、更改、泄露 信息安全属性&#xff1a; CIA三元组&#xff1a;保密性、完整性、可用性 其他属性&#xff1a;真实性、不…
阅读更多...
sql查询分数排名

sql查询分数排名

编写一个 sql 查询来实现分数排名。 如果两个分数相同&#xff0c;则两个分数排名&#xff08;rank&#xff09;相同。请注意&#xff0c;平分后的下一个名次应该是下一个连续的整数值。换句话说&#xff0c;名次之间不应该有“间隔”。 ----------- | id | score | ---------…
阅读更多...
前端---表格标签

前端---表格标签

1. 表格的结构 表格是由行和列组成&#xff0c;好比一个excel文件 2. 表格标签 <table>标签&#xff1a;表示一个表格 <tr>标签&#xff1a;表示表格中的一行 <td>标签&#xff1a;表示表格中的列<th>标签&#xff1a;表示表格中的表头 示例代码: &l…
阅读更多...
鸿蒙开发中的坑（持续更新……）

鸿蒙开发中的坑（持续更新……）

最近在使用鸿蒙开发时&#xff0c;碰到了一些坑&#xff0c;特做记录&#xff0c;如&#xff1a;鸿蒙的preview不能预览&#xff0c;轮播图组件Swiper使用时的问题&#xff0c;console.log() 打印的内容 一、鸿蒙的preview不能预览 首先&#xff0c;只有 ets文件才能预览。 其…
阅读更多...
Spring系列学习二、Spring框架的环境配置

Spring系列学习二、Spring框架的环境配置

Spring框架的环境配置 一、Java环境配置二、 Spring框架的安装与配置三、Maven与Gradle环境的配置四、IDE环境配置&#xff08;Eclipse与IntelliJ IDEA&#xff09;五、结语 一、Java环境配置 所有编程旅程总是得从基础开始&#xff0c;如同乐高积木大作的基座&#xff0c;首先…
阅读更多...
嵌入式——RTC闹钟Alarm

嵌入式——RTC闹钟Alarm

开发流程 配置RTC时钟设置RTC闹钟配置RTC闹钟中断实现中断函数RTC闹钟初始化 // 闹钟外部中断 exti_flag_clear(EXTI_17); exti_init(EXTI_17,EXTI_INTERRUPT,EXTI_TRIG_RISING);// 重置闹钟 rtc_alarm_disable(RTC_ALARM0);rtc_alarm_struct ras; ras.alarm_mask = RTC_ALARM…
阅读更多...
408数据结构错题知识点拾遗

408数据结构错题知识点拾遗

408相关&#xff1a; 408数据结构错题知识点拾遗 408计算机网络错题知识点拾遗 对于数据结构的学习&#xff0c;个人认为要对概念性的东西进行理解&#xff0c;特别是树的性质、图的相关性质和考察的相应算法。应用题强化的话&#xff0c;对于每一章节尾的应用小节&#xff0c…
阅读更多...
使用C++ 标准库map关联式容器根绝键值查找文件是否存在

使用C++ 标准库map关联式容器根绝键值查找文件是否存在

下面这段代码创建了一个简单的示例&#xff0c;演示了如何使用 std::map 存储和检索 std::shared_ptr 类型的对象。代码中的注释已经对每一步进行了说明 #include <iostream> #include <map> #include <string> #include <memory>class ObjFile{ publ…
阅读更多...
PHP函数学习总结

PHP函数学习总结

version_compare&#xff08;比较php版本&#xff09; 用法&#xff1a; version_compare(string $version1, string $version2, ?string $operator null): int|bool//示例 $result version_compare(PHP_VERSION, 8.0.0) > 0 ? ok : fail;echo $result;// 输出ok证明当…
阅读更多...
【Log4j2】Log4j2最佳实践：Log4j2配置超过7天压缩，超过3个月删除文件的滚动日志，分别定义info文件和error文件，按照每小时存储

【Log4j2】Log4j2最佳实践：Log4j2配置超过7天压缩，超过3个月删除文件的滚动日志，分别定义info文件和error文件，按照每小时存储

目录 Log4j2配置 springboot多环境日志配置 参考资料 Log4j2配置 如果你想要在控制台输出美化的日志信息&#xff0c;你可以使用Log4j2的ConsoleAppender和AnsiColorConverter来实现。下面是相应的配置示例&#xff1a; <Configuration status"WARN"><…
阅读更多...
蓝桥杯c/c++程序设计——冶炼金属

蓝桥杯c/c++程序设计——冶炼金属

冶炼金属 问题描述 小蓝有一个神奇的炉子用于将普通金属 O 冶炼成为一种特殊金属 X。这个炉子有一个称作转换率的属性 V&#xff0c;V 是一个正整数&#xff0c;这意味着消耗 V 个普通金属 O 恰好可以冶炼出一个特殊金属 X&#xff0c;当普通金属 O 的数目不足 V 时&#xff0…
阅读更多...
最新文章

Copyright @ 2022~2023