kafka 消息能過濾嗎

是的，Kafka 消息可以進行過濾。在 Kafka 消費者應用程序中，你可以使用消息過濾器來選擇性地消費某些特定的消息。這可以通過編寫自定義的消費者邏輯來實現，例如使用 Kafka 消費者 API 的 poll() 方法來獲取消息，然后根據消息的鍵（key）或值（value）進行過濾。

以下是一個簡單的 Java 示例，展示了如何在 Kafka 消費者中使用消息過濾器：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class FilteredKafkaConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "filtered-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("my-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 過濾條件：只消費鍵為 "filtered-key" 的消息
                if (record.key().equals("filtered-key")) {
                    System.out.printf("Consumed record with key '%s', value '%s', partition %d, offset %d%n",
                            record.key(), record.value(), record.partition(), record.offset());
                }
            }
        }
    }
}

在這個示例中，我們創建了一個 Kafka 消費者，訂閱了名為 “my-topic” 的主題。然后，我們使用 poll() 方法獲取消息，并使用一個簡單的條件來過濾鍵為 “filtered-key” 的消息。如果消息滿足過濾條件，我們將其打印出來。你可以根據需要修改過濾條件。