pc网站如何做移动网站,网站标题被别人改了 应该怎么办,wordpress 后台 shell,云南建站1 kafka消息压缩
kafka关于消息压缩的定义#xff08;来源于官网#xff09;#xff1a; 此为 Kafka 中端到端的块压缩功能。如果启用#xff0c;数据将由 producer 压缩#xff0c;以压缩格式写入服务器#xff0c;并由 consumer 解压缩。压缩将提高 consumer 的吞吐量…1 kafka消息压缩
kafka关于消息压缩的定义来源于官网 此为 Kafka 中端到端的块压缩功能。如果启用数据将由 producer 压缩以压缩格式写入服务器并由 consumer 解压缩。压缩将提高 consumer 的吞吐量但需付出一定的解压成本。 压缩就是用时间换空间其基本理念是基于重复将重复的片段编码为字典字典的 key 为重复片段value 为更短的代码比如序列号然后将原始内容中的片段用代码表示达到缩短内容的效果压缩后的内容则由字典和代码序列两部分组成。解压时根据字典和代码序列可无损地还原为原始内容。通常来讲重复越多压缩效果越好。比如 JSON 是 Kafka 消息中常用的序列化格式单条消息内可能并没有多少重复片段但如果是批量消息则会有大量重复的字段名批量中消息越多则重复越多这也是为什么 Kafka 更偏向块压缩而不是单条消息压缩。
2 kafka的消息压缩类型对比 目前 Kafka 共支持四种主要的压缩类型Gzip、Snappy、Lz4 和 Zstd。关于这几种压缩的特性。
压缩类型压缩率CPU 使用率压缩速度带宽使用率GzipHighestHighestSlowestLowestSnappyMediumModerateModerateMediumLz4LowLowestFastestHighestZstdMediumModerateModerateMedium 从上表可知Snappy 在 CPU 使用率、压缩比、压缩速度和网络带宽使用率之间实现良好的平衡我们最终也是采用的该类型进行压缩试点。这里值得一提的是Zstd 是 Facebook 于 2016 年开源的新压缩算法压缩率和压缩性能都不错具有与 SnappyGoogle 杰作相似的特性直到 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.typesnappy
也可以通过 configs 命令修改 topic 的 compression.type 取值命令如下
bin/kafka-configs.sh \
--entity-type topics \
--entity-name my-topic \
--alter \
--add-config compression.typesnappy
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);KafkaProducerString, 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 imsg;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自动解压缩。
