JavaでKafkaフィルタ処理を行う方法は？

2年 ago
海斗, 葵
2 minutes

JavaではKafkaのConsumer APIを使用してメッセージをフィルタリングすることができます。Consumer APIはメッセージのキー値、パーティション、オフセットなどの属性に基づいてメッセージをフィルタリングするための柔軟な方法を提供します。

一般的に使用されているフィルタリング方法を以下に示します：

  1. キー値フィルタはメッセージをコンシューマレコードのキー値を設定することでフィルタできます。指定のトピックを購読するには、コンシューマAPIのsubscribe()メソッドを使用し、コンシューマのキー値フィルター条件を指定するには、ConsumerRebalanceListenerのonPartitionsAssigned()メソッドを設定します。
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");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("topic1"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for (TopicPartition partition : partitions) {
            // 设置键值过滤条件
            consumer.seek(partition, 0);
        }
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 撤销键值过滤条件
    }
});
  1. 区割りでフィルタリングする：ConsumerRebalanceListener の onPartitionsAssigned() メソッドを設定することで、消費者の区割りフィルタリング条件を指定することができます。
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");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("topic1"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for (TopicPartition partition : partitions) {
            if (partition.partition() == 1) {
                // 过滤指定分区
                consumer.seek(partition, 0);
            }
        }
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 撤销分区过滤条件
    }
});
  1. オフセットフィルタ:ConsumerRebalanceListenerのonPartitionsAssigned()メソッドを設定することで、消費者のオフセットフィルタ条件を指定できる。
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");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("topic1"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for (TopicPartition partition : partitions) {
            // 设置偏移量过滤条件
            consumer.seek(partition, 10);
        }
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 撤销偏移量过滤条件
    }
});

上記の方法で、Kafkaメッセージのフィルタリングを実現できます。具体的なニーズに応じて、適切なフィルタリング方法を選択できます。

bannerAds