Kafka-Stream メッセージストリーミングプロセスと原則の簡単な分析

2024-07-12

以下はケースの組み合わせです: Kafka メッセージストリーミング処理の実行プロセスをテストおよび説明するために、メッセージ内の単語の出現数をカウントします。

Maven の依存関係

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <exclusions>
                <exclusion>
                    <artifactId>connect-json</artifactId>
                    <groupId>org.apache.kafka</groupId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka-clients</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </dependency>
    </dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

準備

まず、メッセージプロデューサ、メッセージコンシューマ、およびストリームプロセッサとして 3 つのクラスをそれぞれ記述して作成します。
KafkaStreamProducer:メッセージプロデューサー

public class KafkaStreamProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        //kafka的连接地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.246.128:9092");
        //发送失败，失败的重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG, 5);
        //消息key的序列化器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //消息value的序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        for (int i = 0; i < 5; i++) {
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("kafka-stream-topic-input", "hello kafka");
            producer.send(producerRecord);
        }

        producer.close();

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

メッセージプロデューサーはトピックに報告しますkafka-stream-topic-input5回送信hello kafka
KafkaStreamConsumer:メッセージコンシューマ

public class KafkaStreamConsumer {
    public static void main(String[] args) {
        Properties properties = new Properties();
        //kafka的连接地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.246.128:9092");
        //消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
        //消息的反序列化器
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        //手动提交偏移量
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        //订阅主题
        consumer.subscribe(Collections.singletonList("kafka-stream-topic-output"));

        try {
            while (true) {
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    System.out.println("consumerRecord.key() = " + consumerRecord.key());
                    System.out.println("consumerRecord.value() = " + consumerRecord.value());
                }
                // 异步提交偏移量
                consumer.commitAsync();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 同步提交偏移量
            consumer.commitSync();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

KafkaStreamQuickStart：ストリーミング処理クラス

public class KafkaStreamQuickStart {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.246.128:9092");
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-quickstart");

        StreamsBuilder streamsBuilder = new StreamsBuilder();

        //流式计算
        streamProcessor(streamsBuilder);

        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), properties);

        kafkaStreams.start();
    }

    /**
     * 消息格式：hello world hello world
     * 配置并处理流数据。
     * 使用StreamsBuilder创建并配置KStream，对输入的主题中的数据进行处理，然后将处理结果发送到输出主题。
     * 具体处理包括：分割每个消息的值，按值分组，对每个分组在10秒的时间窗口内进行计数，然后将结果转换为KeyValue对并发送到输出主题。
     *
     * @param streamsBuilder 用于构建KStream对象的StreamsBuilder。
     */
    private static void streamProcessor(StreamsBuilder streamsBuilder) {
        // 从"kafka-stream-topic-input"主题中读取数据流
        KStream<String, String> stream = streamsBuilder.stream("kafka-stream-topic-input");
        System.out.println("stream = " + stream);
        // 将每个值按空格分割成数组，并将数组转换为列表，以扩展单个消息的值
        stream.flatMapValues((ValueMapper<String, Iterable<String>>) value -> {
                    String[] valAry = value.split(" ");
                    return Arrays.asList(valAry);
                })
                // 按消息的值进行分组，为后续的窗口化计数操作做准备
                .groupBy((key, value) -> value)
                // 定义10秒的时间窗口，在每个窗口内对每个分组进行计数
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .count()
                // 将计数结果转换为流，以便进行进一步的处理和转换
                .toStream()
                // 显示键值对的内容，并将键和值转换为字符串格式
                .map((key, value) -> {
                    System.out.println("key = " + key);
                    System.out.println("value = " + value);
                    return new KeyValue<>(key.key().toString(), value.toString());
                })
                // 将处理后的流数据发送到"kafka-stream-topic-output"主题
                .to("kafka-stream-topic-output");
    }
    
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

この処理クラスはまずトピックから始まりますkafka-stream-topic-inputからメッセージデータを取得し、処理後にトピックに送信しますkafka-stream-topic-output、次にメッセージコンシューマKafkaStreamConsumer消費する

の結果

ここに画像の説明を挿入します

ストリーミング処理のプロセスと原理の説明

初期段階

からトピックに入る場合kafka-stream-topic-inputデータストリームを読み取る場合、各メッセージはキーと値のペアになります。入力メッセージのキーがnullメッセージが入力トピックに送信される方法に応じて、または特定の文字列。

KStream<String, String> stream = streamsBuilder.stream("kafka-stream-topic-input");
1

分割メッセージ値

使用flatMapValuesこのメソッドはメッセージの値を分割しますが、この操作ではメッセージのキーは変更されません。メッセージを入力するキーがnull、この段階ではメッセージのキーはまだnull。

stream.flatMapValues((ValueMapper<String, Iterable<String>>) value -> {
    String[] valAry = value.split(" ");
    return Arrays.asList(valAry);
})
1
2
3
4

メッセージ値によるグループ化

Kafka ストリームで使用する場合groupByメソッドがストリームをグループ化するとき、実際には新しいキーを指定しており、そのキーは後続のウィンドウ操作や集計操作に使用されます。この場合groupByメッセージを値ごとにグループ化するためにメソッドが使用されます。

.groupBy((key, value) -> value)
1

これは、グループ化操作の後、ストリーム内の各メッセージのキーがメッセージの値に設定されることを意味します。したがって、フォローアップするときは、mapメソッドで見るkeyパラメータ、これkey実際にはメッセージの元の値です。groupByその後、メッセージの値がキーになります。

時間枠とカウントを定義する

この段階では、メッセージがウィンドウ化されてカウントされますが、キーは変更されません。

.windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
.count()
1
2

カウント結果をストリームに変換する

カウント結果をストリームに変換するとき、グループ化するときのキーは以前と同じままです。

.toStream()
1

結果の処理と変換

存在するmap方法、わかりますかkeyこのパラメータは実際にはグループ化されたキーであり、メッセージの元の値です。

.map((key, value) -> {
    System.out.println("key = " + key);
    System.out.println("value = " + value);
    return new KeyValue<>(key.key().toString(), value.toString());
})
1
2
3
4
5

mapメソッド内key.key().toString()キーの文字列表現を取得することですが、value.toString()カウント値を文字列に変換することです。

処理されたデータを出力トピックに送信する

.to("kafka-stream-topic-output");
1

技術共有