Краткий анализ процесса и принципов потоковой передачи сообщений Kafka-Stream

Краткий анализ процесса и принципов потоковой передачи сообщений 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

Подготовка

Сначала напишите и создайте три класса, соответственно, в качестве производителей сообщений, потребителей сообщений и потоковых процессоров.
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-inputОтправлено пять раз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 Streams при использовании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

Обмен технологиями