掌握Kafka Streams：自定义时间戳提取器与窗口操作深度解析

import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.TimestampExtractor;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.Properties;

public class KafkaStreamsApp {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-streams-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // ... 其他配置，如默认的键值序列化器

        // 定义一个自定义时间戳提取器
        props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, MyEventTimeExtractor.class.getName());

        // KafkaStreams streams = new KafkaStreams(builder.build(), props);
        // streams.start();
    }

    // 自定义时间戳提取器的实现示例
    public static class MyEventTimeExtractor implements TimestampExtractor {
        @Override
        public long extract(ConsumerRecord<Object, Object> record, long previousTimestamp) {
            // 假设事件时间存储在记录的值中，例如是一个JSON字符串，包含"eventTime"字段
            // 实际应用中，需要根据具体的数据格式（如JSON, Avro等）进行解析
            if (record.value() instanceof String) {
                String value = (String) record.value();
                try {
                    // 假设值是简单的长整型字符串表示时间戳
                    return Long.parseLong(value);
                } catch (NumberFormatException e) {
                    // 如果解析失败，可以返回记录的默认时间戳或上一个时间戳
                    return record.timestamp() > 0 ? record.timestamp() : previousTimestamp;
                }
            }
            // 如果值不是字符串，或者无法解析，则回退到记录的默认时间戳或上一个时间戳
            return record.timestamp() > 0 ? record.timestamp() : previousTimestamp;
        }
    }
}

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import java.time.Duration;
import java.util.Properties;

public class TumblingWindowExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "tumbling-window-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, KafkaStreamsApp.MyEventTimeExtractor.class.getName());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> sourceStream = builder.stream("input-topic");

        sourceStream
            .groupByKey() // 根据键进行分组，所有具有相同键的记录将被聚合
            .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).grace(Duration.ofSeconds(30))) // 5分钟翻滚窗口，30秒宽限期
            .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("tumbling-window-counts")
                    .withKeySerde(Serdes.String())
                    .withValueSerde(Serdes.Long())) // 计数聚合
            .toStream()
            .to("output-topic", Produced.with(WindowedSerdes.timeWindowedSerdeFrom(String.class), Serdes.Long()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        // 添加关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}