Kafka如何配置消息压缩

Apache Kafka 支持多种消息压缩算法，包括 gzip、snappy、lz4 和 zstd。要配置 Kafka 消息压缩，需要在 broker 和 producer 端进行设置。以下是配置消息压缩的步骤：

Broker 配置

编辑 server.properties 文件：打开 Kafka 安装目录下的 config/server.properties 文件。

设置压缩类型：在 server.properties 文件中，找到或添加以下配置项，并设置为你想要的压缩算法：

# 启用压缩
compression.type=gzip  # 可选值：gzip, snappy, lz4, zstd

# 设置压缩级别（仅对 gzip 有效）
compression.gzip.level=9

重启 Kafka Broker：修改配置后，重启 Kafka Broker 以使更改生效。

bin/kafka-server-stop.sh
bin/kafka-server-start.sh config/server.properties

Producer 配置

编辑 Producer 配置文件：打开你的 Producer 应用程序的配置文件，或者直接在代码中设置 Producer 配置。

设置压缩类型：在 Producer 配置中，找到或添加以下配置项，并设置为你想要的压缩算法：

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("compression.type", "gzip");  // 可选值：gzip, snappy, lz4, zstd

创建 Producer 实例：使用配置好的属性创建 Kafka Producer 实例。

KafkaProducer producer = new KafkaProducer<>(props);

Consumer 配置

通常情况下，Consumer 不需要特别配置压缩类型，因为 Kafka 会自动解压缩接收到的消息。但是，如果你需要手动处理压缩消息，可以在 Consumer 配置中设置相应的解压缩器。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
KafkaConsumer consumer = new KafkaConsumer<>(props);

通过以上步骤，你可以在 Kafka 中配置消息压缩，从而提高数据传输效率和存储利用率。