Kafka如何配置消息壓縮

Apache Kafka 支持多種消息壓縮算法，包括 gzip、snappy、lz4 和 zstd。要配置 Kafka 消息壓縮，需要在 broker 和 producer 端進行設置。以下是配置消息壓縮的步驟：

Broker 配置

1. 編輯 server.properties 文件： 打開 Kafka 安裝目錄下的 config/server.properties 文件。

2. 設置壓縮類型： 在 server.properties 文件中，找到或添加以下配置項，并設置為你想要的壓縮算法：

   # 啟用壓縮
   compression.type=gzip  # 可選值：gzip, snappy, lz4, zstd
   
   # 設置壓縮級別（僅對 gzip 有效）
   compression.gzip.level=9
   

   • compression.type：指定默認的壓縮算法?？梢栽O置為 gzip, snappy, lz4, 或 zstd。
   • compression.gzip.level：僅對 gzip 壓縮有效，設置壓縮級別（1-9），9 是最高壓縮級別。

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

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

Producer 配置

1. 編輯 Producer 配置文件： 打開你的 Producer 應用程序的配置文件，或者直接在代碼中設置 Producer 配置。

2. 設置壓縮類型： 在 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
   

   • compression.type：指定 Producer 發送消息時使用的壓縮算法。

3. 創建 Producer 實例： 使用配置好的屬性創建 Kafka Producer 實例。

   KafkaProducer<String, String> 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<String, String> consumer = new KafkaConsumer<>(props);

通過以上步驟，你可以在 Kafka 中配置消息壓縮，從而提高數據傳輸效率和存儲利用率。