kafka关于消息压缩的定义(来源于官网):
????????此为 Kafka 中端到端的块压缩功能。如果启用,数据将由 producer 压缩,以压缩格式写入服务器,并由 consumer 解压缩。压缩将提高 consumer 的吞吐量,但需付出一定的解压成本。
????????压缩就是用时间换空间,其基本理念是基于重复,将重复的片段编码为字典,字典的 key 为重复片段,value 为更短的代码,比如序列号,然后将原始内容中的片段用代码表示,达到缩短内容的效果,压缩后的内容则由字典和代码序列两部分组成。解压时根据字典和代码序列可无损地还原为原始内容。通常来讲,重复越多,压缩效果越好。比如 JSON 是 Kafka 消息中常用的序列化格式,单条消息内可能并没有多少重复片段,但如果是批量消息,则会有大量重复的字段名,批量中消息越多,则重复越多,这也是为什么 Kafka 更偏向块压缩,而不是单条消息压缩。
????????目前 Kafka 共支持四种主要的压缩类型:Gzip、Snappy、Lz4 和 Zstd。关于这几种压缩的特性。
压缩类型 | 压缩率 | CPU 使用率 | 压缩速度 | 带宽使用率 |
---|---|---|---|---|
Gzip | Highest | Highest | Slowest | Lowest |
Snappy | Medium | Moderate | Moderate | Medium |
Lz4 | Low | Lowest | Fastest | Highest |
Zstd | Medium | Moderate | Moderate | Medium |
????????从上表可知,Snappy 在 CPU 使用率、压缩比、压缩速度和网络带宽使用率之间实现良好的平衡,我们最终也是采用的该类型进行压缩试点。这里值得一提的是,Zstd 是 Facebook 于 2016 年开源的新压缩算法,压缩率和压缩性能都不错,具有与 Snappy(Google 杰作)相似的特性,直到 Kafka 的 2.1.0 版本才引入支持。
????????压缩是需要额外的 CPU 代价的,并且会带来一定的消息分发延迟,因而在压缩前要慎重考虑是否有必要。
????????Kafka 通过配置属性?compression.type
?控制是否压缩。该属性在 producer 端和 broker 端各自都有一份,也就是说,我们可以选择在 producer 或 broker 端开启压缩,对应的应用场景各有不同。目前没有尝试在broker段开启压缩。
????????Broker 端的?compression.type
?属性默认值为?producer
,即直接继承 producer 端所发来消息的压缩方式,无论消息采用何种压缩或者不压缩,broker 都原样存储。、
????????在 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
????????跟 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; // 设置字符串大小为2MB
int 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。
可能发生解压的地方依然是两处: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自动解压缩。