// 1. 创建状态存储
StoreBuilder> storeBuilder =
Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore(“fgedu-state-store”),
Serdes.String(),
Serdes.Long()
);

// 2. 添加状态存储到拓扑
builder.addStateStore(storeBuilder);

// 3. 在Processor中使用状态存储
KStream processed = source
.process(() -> new Processor() {
private KeyValueStore stateStore;

@Override
public void init(ProcessorContext context) {
this.stateStore = context.getStateStore(“fgedu-state-store”);
}

@Override
public void process(String key, String value) {
Long count = stateStore.get(key);
if (count == null) {
count = 0L;
}
stateStore.put(key, count + 1);
}

@Override
public void close() {}
}, “fgedu-state-store”);

# 状态存储类型
– RocksDB: 持久化存储，适合大数据量
– InMemory: 内存存储，适合小数据量
– Persistent: 持久化存储
– Caching: 缓存存储

// 1. 配置应用服务器
props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, “192.168.1.51:8080”);

// 2. 创建可查询的状态存储
KTable counts = source
.groupBy((key, value) -> value)
.count(Materialized.as(“fgedu-counts-store”));

// 3. 查询状态存储
ReadOnlyKeyValueStore store =
streams.store(“fgedu-counts-store”, QueryableStoreTypes.keyValueStore());

// 4. 查询特定Key的值
Long count = store.get(“hello”);

// 5. 查询所有Key
KeyValueIterator all = store.all();
while (all.hasNext()) {
KeyValue next = all.next();
System.out.println(next.key + “: ” + next.value);
}

// 6. 范围查询
KeyValueIterator range = store.range(“a”, “z”);

package com.fgedu.kafka;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.common.serialization.Serdes;
import java.util.Arrays;
import java.util.Properties;
import java.util.regex.Pattern;

public class WordCountDemo {
public static void main(String[] args) {
// 配置
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, “fgedu-wordcount”);
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, “192.168.1.51:9092”);
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

// 构建拓扑
StreamsBuilder builder = new StreamsBuilder();

// 从Topic读取文本
KStream textLines = builder.stream(“fgedu-text-input”);

// 分词并统计
Pattern pattern = Pattern.compile(“\\W+”, Pattern.UNICODE_CHARACTER_CLASS);
KTable wordCounts = textLines
.flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase())))
.groupBy((key, word) -> word)
.count();

// 输出结果
wordCounts.toStream().to(
“fgedu-wordcount-output”,
Produced.with(Serdes.String(), Serdes.Long())
);

// 启动应用
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}
}

# 创建Topic
$ /bigdata/app/kafka/bin/kafka-topics.sh –create \
–topic fgedu-text-input \
–partitions 3 \
–replication-factor 3 \
–bootstrap-server 192.168.1.51:9092

$ /bigdata/app/kafka/bin/kafka-topics.sh –create \
–topic fgedu-wordcount-output \
–partitions 3 \
–replication-factor 3 \
–bootstrap-server 192.168.1.51:9092

# 运行应用并发送测试数据
$ /bigdata/app/kafka/bin/kafka-console-producer.sh \
–topic fgedu-text-input \
–bootstrap-server 192.168.1.51:9092

>hello world hello kafka
>kafka streams kafka demo
>fgedu kafka fgedu demo

# 查看统计结果
$ /bigdata/app/kafka/bin/kafka-console-consumer.sh \
–topic fgedu-wordcount-output \
–from-beginning \
–property print.key=true \
–property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
–property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer \
–bootstrap-server 192.168.1.51:9092

demo 2
fgedu 2
hello 2
kafka 4
streams 1
world 1

// 订单流
KStream orders = builder.stream(“fgedu-orders”);

// 支付流
KStream payments = builder.stream(“fgedu-payments”);

// 连接两个流
KStream joined = orders
.join(payments,
(order, payment) -> new OrderPayment(order, payment),
JoinWindows.of(Duration.ofMinutes(5)),
StreamJoined.with(
Serdes.String(),
new JsonSerde<>(Order.class),
new JsonSerde<>(Payment.class)
)
);

// 输出连接结果
joined.to(“fgedu-order-payment”);

# 创建Topic
$ /bigdata/app/kafka/bin/kafka-topics.sh –create \
–topic fgedu-orders \
–partitions 3 \
–replication-factor 3 \
–bootstrap-server 192.168.1.51:9092

$ /bigdata/app/kafka/bin/kafka-topics.sh –create \
–topic fgedu-payments \
–partitions 3 \
–replication-factor 3 \
–bootstrap-server 192.168.1.51:9092

$ /bigdata/app/kafka/bin/kafka-topics.sh –create \
–topic fgedu-order-payment \
–partitions 3 \
–replication-factor 3 \
–bootstrap-server 192.168.1.51:9092

# 排查步骤
# 1. 检查配置
$ grep -E “application.id|bootstrap.servers” streams.properties

application.id=fgedu-streams-demo
bootstrap.servers=192.168.1.51:9092

# 2. 检查Topic是否存在
$ /bigdata/app/kafka/bin/kafka-topics.sh –list \
–bootstrap-server 192.168.1.51:9092

# 3. 检查状态目录权限
$ ls -la /bigdata/streams/state

# 4. 检查日志
$ tail -100 logs/streams.log

# 常见错误
# 1. application.id冲突
# 解决：使用唯一的application.id

# 2. Topic不存在
# 解决：先创建Topic

# 3. 状态目录权限问题
# 解决：chmod 755 /bigdata/streams/state

# 排查步骤
# 1. 检查消费者延迟
$ /bigdata/app/kafka/bin/kafka-consumer-groups.sh –describe \
–group fgedu-streams-demo \
–bootstrap-server 192.168.1.51:9092

# 2. 检查线程数
# 增加num.stream.threads

# 3. 检查状态存储大小
$ du -sh /bigdata/streams/state

# 优化方案
# 1. 增加处理线程
num.stream.threads=4

# 2. 增加缓存大小
cache.max.bytes.buffering=52428800

# 3. 调整提交间隔
commit.interval.ms=100

# 4. 增加实例数量
# 部署更多应用实例

# 部署最佳实践
1. 监控消费者延迟
2. 监控状态存储大小
3. 配置合理的资源
4. 实现健康检查
5. 制定故障恢复预案

# 性能优化
1. 增加处理线程
2. 增加缓存大小
3. 优化状态存储
4. 调整提交间隔