1
// KStream：每条记录都是独立事件
2
// 订单事件流：order-1 创建, order-1 支付, order-1 发货
3
KStream<String, OrderEvent> orderEvents = builder.stream("orders");
4
// 输出：order-1 → {type: CREATED}
5
//       order-1 → {type: PAID}
6
//       order-1 → {type: SHIPPED}
7

8
// KTable：相同 Key 的记录是更新
9
// 订单最新状态：order-1 的状态不断更新
10
KTable<String, OrderState> orderStates = builder.table("order-states");
11
// 输出：order-1 → {status: CREATED}
12
//       order-1 → {status: PAID}      (覆盖 CREATED)
13
//       order-1 → {status: SHIPPED}   (覆盖 PAID)

1
// 流 → 表：订单金额汇总
2
KStream<String, Order> orders = builder.stream("orders");
3

4
KTable<String, Double> customerTotals = orders
5
    .groupBy((key, order) -> KeyValue.pair(order.getCustomerId(), order.getAmount()))
6
    .aggregate(
7
        () -> 0.0,                                          // 初始值
8
        (customerId, amount, total) -> total + amount,      // 累加器
9
        Materialized.with(Serdes.String(), Serdes.Double()) // 状态存储
10
    );
11

12
// 表 → 流：金额变更事件
13
KStream<String, Double> totalChanges = customerTotals.toStream();
14
totalChanges.to("customer-total-changes");

1
// 翻滚窗口：每 5 分钟统计订单量
2
KStream<String, Order> orders = builder.stream("orders");
3

4
KTable<Windowed<String>, Long> orderCounts = orders
5
    .groupByKey()
6
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
7
    .count(Materialized.as("order-counts-store"));
8

9
// 跳跃窗口：每 1 分钟滑动，窗口大小 5 分钟
10
KTable<Windowed<String>, Double> movingAvg = orders
11
    .groupByKey()
12
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5))
13
        .advanceBy(Duration.ofMinutes(1)))
14
    .aggregate(
15
        () -> 0.0,
16
        (key, order, aggregate) -> aggregate + order.getAmount(),
17
        Materialized.with(Serdes.String(), Serdes.Double())
18
    );
19

20
// 会话窗口：30 分钟不活跃则关闭会话
21
KTable<Windowed<String>, Integer> sessionCounts = orders
22
    .groupByKey()
23
    .windowedBy(SessionWindows.with(Duration.ofMinutes(30)))
24
    .aggregate(
25
        () -> 0,
26
        (key, order, count) -> count + 1,
27
        (aggKey, leftAgg, rightAgg) -> leftAgg + rightAgg,
28
        Materialized.with(Serdes.String(), Serdes.Integer())
29
    );

1
// 使用事件时间（推荐）
2
// 通过 TimestampExtractor 提取事件中的时间戳
3
public class OrderTimestampExtractor implements TimestampExtractor {
4
    @Override
5
    public long extract(ConsumerRecord<Object, Object> record, long partitionTime) {
6
        Order order = (Order) record.value();
7
        return order.getEventTime();  // 使用事件中的时间戳
8
    }
9
}
10

11
// 配置
12
props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
13
    OrderTimestampExtractor.class.getName());
14

15
// 处理迟到数据
16
// 设置允许的迟到时间
17
.windowedBy(TimeWindows.of(Duration.ofMinutes(5))
18
    .grace(Duration.ofMinutes(1)))  // 允许 1 分钟迟到

1
// 状态存储配置
2
StoreBuilder<KeyValueStore<String, Long>> countStore = Stores
3
    .keyValueStoreBuilder(
4
        Stores.persistentKeyValueStore("order-counts"),  // RocksDB 持久化
5
        Serdes.String(),
6
        Serdes.Long()
7
    )
8
    .withCachingEnabled()     // 启用缓存（减少 I/O）
9
    .withLoggingEnabled();    // 启用 Changelog（容错）
10

11
builder.addStateStore(countStore);
12

13
// 在 Processor 中使用状态存储
14
public class OrderProcessor implements Processor<String, Order, String, Order> {
15
    private KeyValueStore<String, Long> countStore;
16

17
    @Override
18
    public void init(ProcessorContext<String, Order> context) {
19
        this.countStore = context.getStateStore("order-counts");
20
    }
21

22
    @Override
23
    public void process(Record<String, Order> record) {
24
        String customerId = record.value().getCustomerId();
25
        Long count = countStore.get(customerId);
26
        countStore.put(customerId, count == null ? 1 : count + 1);
27
    }
28
}

1
# 查看 Changelog Topic
2
kafka-topics --list --bootstrap-server localhost:9092 | grep changelog
3
# 输出：order-counts-store-changelog
4
#        customer-totals-store-changelog
5

6
# Changelog Topic 配置
7
# cleanup.policy=compact  (保留最新值)
8
# segment.bytes=67108864  (64MB)
9
# retention.bytes=-1      (无大小限制)

1
// 使用 DSL 构建拓扑
2
StreamsBuilder builder = new StreamsBuilder();
3

4
// Source: 从 Kafka 读取
5
KStream<String, Order> orders = builder.stream("orders");
6

7
// Process: 过滤无效订单
8
KStream<String, Order> validOrders = orders
9
    .filter((key, order) -> order.getAmount() > 0);
10

11
// Transform: 按客户分组
12
KGroupedStream<String, Order> grouped = validOrders
13
    .groupBy((key, order) -> KeyValue.pair(order.getCustomerId(), order));
14

15
// Aggregate: 计算客户订单总额
16
KTable<String, Double> totals = grouped
17
    .aggregate(
18
        () -> 0.0,
19
        (key, order, total) -> total + order.getAmount(),
20
        Materialized.with(Serdes.String(), Serdes.Double())
21
    );
22

23
// Sink: 写回 Kafka
24
totals.toStream().to("customer-totals");
25

26
// 查看拓扑描述
27
Topology topology = builder.build();
28
System.out.println(topology.describe());

1
// KStream-KTable Join：订单 + 客户信息
2
KStream<String, Order> orders = builder.stream("orders");
3
KTable<String, Customer> customers = builder.table("customers");
4

5
KStream<String, EnrichedOrder> enriched = orders.join(
6
    customers,
7
    (order, customer) -> new EnrichedOrder(order, customer)
8
);
9

10
// KStream-KStream Join：订单 + 支付
11
KStream<String, Order> orders = builder.stream("orders");
12
KStream<String, Payment> payments = builder.stream("payments");
13

14
KStream<String, OrderPayment> joined = orders.join(
15
    payments,
16
    (order, payment) -> new OrderPayment(order, payment),
17
    JoinWindows.of(Duration.ofMinutes(5))  // 5 分钟内匹配
18
);

1
# Kafka Streams 配置
2
# 线程数（每个实例）
3
num.stream.threads=4
4

5
# 状态目录
6
state.dir=/var/lib/kafka-streams
7

8
# exactly-once
9
processing.guarantee=exactly_once_v2
10

11
# 缓存大小
12
cache.max.bytes.buffering=10485760  # 10MB
13

14
# 提交间隔
15
commit.interval.ms=1000
16

17
# 应用 ID（唯一标识）
18
application.id=order-streams-app

1
# 查看 Streams 应用状态
2
kafka-consumer-groups --bootstrap-server localhost:9092 \
3
    --describe --group order-streams-app
4

5
# JMX 监控
6
# kafka.streams:type=stream-task-metrics,task-id=*
7
# kafka.streams:type=stream-processor-metrics,processor-id=*

1
// 启用交互式查询
2
KTable<String, Double> customerTotals = orders
3
    .groupBy((key, order) -> KeyValue.pair(order.getCustomerId(), order.getAmount()))
4
    .aggregate(
5
        () -> 0.0,
6
        (customerId, amount, total) -> total + amount,
7
        Materialized.as("customer-totals-queryable-store")  // 命名存储，可查询
8
    );
9

10
// 查询本地状态存储
11
ReadOnlyKeyValueStore<String, Double> store =
12
    kafkaStreams.store("customer-totals-queryable-store", QueryableStoreTypes.keyValueStore());
13

14
// 按 Key 查询
15
Double total = store.get("customer-123");
16

17
// 范围查询
18
KeyValueIterator<String, Double> range = store.range("customer-100", "customer-200");

1
// 查找 Key 所在的实例
2
StreamsMetadata metadata = kafkaStreams.streamsMetadataForStore("customer-totals-queryable-store", "customer-123");
3
String host = metadata.host();
4
int port = metadata.port();
5

6
// 通过 HTTP 转发查询请求到正确的实例
7
// GET http://{host}:{port}/state/customer-totals-queryable-store/customer-123

1
StreamsBuilder builder = new StreamsBuilder();
2

3
// 1. 读取订单事件流
4
KStream<String, OrderEvent> orders = builder.stream("orders");
5

6
// 2. 过滤有效订单
7
KStream<String, OrderEvent> validOrders = orders
8
    .filter((key, event) -> event.getAmount() != null && event.getAmount() > 0);
9

10
// 3. 按客户分组，5 分钟翻滚窗口统计
11
KTable<Windowed<String>, OrderSummary> windowedSummary = validOrders
12
    .groupBy((key, event) -> KeyValue.pair(event.getCustomerId(), event))
13
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).grace(Duration.ofMinutes(1)))
14
    .aggregate(
15
        OrderSummary::new,
16
        (customerId, event, summary) -> summary.add(event),
17
        Materialized.with(Serdes.String(), new JsonSerde<>(OrderSummary.class))
18
    );
19

20
// 4. 检测异常：5 分钟内消费超过阈值
21
KStream<String, Alert> alerts = windowedSummary
22
    .toStream()
23
    .filter((windowedKey, summary) -> summary.getTotalAmount() > 10000)
24
    .map((windowedKey, summary) -> new KeyValue<>(
25
        windowedKey.key(),
26
        new Alert("HIGH_SPENDING", windowedKey.key(), summary.getTotalAmount())
27
    ));
28

29
// 5. 输出告警
30
alerts.to("order-alerts", Produced.with(Serdes.String(), new JsonSerde<>(Alert.class)));
31

32
// 6. 物化客户最新状态（可查询）
33
KTable<String, CustomerStats> customerStats = validOrders
34
    .groupBy((key, event) -> KeyValue.pair(event.getCustomerId(), event))
35
    .aggregate(
36
        CustomerStats::new,
37
        (customerId, event, stats) -> stats.update(event),
38
        Materialized.as("customer-stats-store")
39
    );