在 Apache Kafka 中，消息压缩可以通过配置生产者和消费者的压缩类型来启用。Kafka 支持多种压缩算法，包括 none（不压缩）、gzip、snappy、lz4 和 zstd。以下是配置 Kafka 消息压缩的详细步骤：

Kafka Producer 配置

在创建 Kafka Producer 时，可以通过设置 compression.type 属性来选择压缩算法。以下是一个使用 Snappy 压缩算法的示例：

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", "snappy"); // 设置压缩算法为 Snappy

Producer producer = new KafkaProducer<>(props);

Kafka Broker 配置

在 Kafka Broker 的 server.properties 文件中，可以设置默认的压缩算法。以下是一个配置示例：

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

Kafka Consumer 配置

Kafka Consumer 默认会解压缩接收到的消息。如果需要处理压缩消息，可以在创建消费者时设置 enable.auto.commit 为 false，并在消费消息时手动处理压缩数据。以下是一个使用 Snappy 解压缩的示例：

Properties props = new Properties();
props.put("group.id", "test");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");

Consumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));

while (true) {
    ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord record : records) {
        // Snappy 解压缩
        byte[] compressedData = record.value().getBytes();
        byte[] decompressedData = Snappy.uncompress(compressedData);
        String value = new String(decompressedData, StandardCharsets.UTF_8);
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), value);
    }
}

压缩算法的选择和性能权衡

• GZIP：提供较高的压缩率，但压缩和解压缩速度较慢，适合对数据压缩率要求较高而对速度要求不那么严格的场景。
• Snappy：压缩和解压缩速度快，适合需要快速处理数据流的场景，例如实时数据分析。
• LZ4：压缩率适中，但解压缩速度非常快，适合对延迟敏感的应用，如实时通信系统。
• Zstandard (zstd)：提供超高的压缩比，适合存储成本高且对延迟不敏感的场景。

通过以上配置和注意事项，您可以在 Kafka 中启用和优化消息压缩，从而提高数据传输效率和存储利用率。