<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <version>3.6.2</version>
      </dependency>

@Slf4j
@Component
public class MyKafkaStreamProcessor {
	@Value("${spring.kafka.bootstrap-servers}")
	private String bootstrapServers;
	@PostConstruct
	private void init () {
		String appId = "my-kafka-streams-app";
		myKafkaStreams(appId);
		log.info("? Kafka Streams:{} 初始化完成，開始監(jiān)聽 topic: {}", appId, "source-topic");
	}
	public void myKafkaStreams(String appId) {
		/*=======配置=======*/
		Properties config  = new Properties();
		config.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
		config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
		StreamsBuilder builder = new StreamsBuilder();
		/*=======構(gòu)建拓撲結(jié)構(gòu)=======*/
		// 數(shù)據(jù)清洗
		KStream<String, Order> stream = builder
			.stream("source-topic", Consumed.with(Serdes.String(), new JsonSerde<>(Order.class)))
			.mapValues(value -> {
				// do something
				return value;
			});
		// 過濾出從app創(chuàng)建的訂單 并進行處理
		stream.filter((k, v) -> Order.getSource.equals("app"))
			.foreach((k, v) -> {
				// do something
			});
		// 發(fā)送到第一個topic
		stream.mapValues(value -> JSON.toJSONString(value), Named.as("to-the-first-target-topic-processor"))
			.to("the-first-target-topic", Produced.with(Serdes.String(), Serdes.String()));
		// 發(fā)送到第二個topic
		stream.filter((k, v) -> {
				// filter something
			})
			.mapValues(value -> {
				// map to another object
			}, Named.as("to-the-second-target-topic-processor"))
			.to("the-second-target-topic", Produced.with(Serdes.String(), Serdes.String()));
		/*=======創(chuàng)建KafkaStreams=======*/
		KafkaStreams streams = new KafkaStreams(builder.build(), config);
		/*=======設(shè)置異常處理器=======*/
		streams.setUncaughtExceptionHandler(new CustomStreamsUncaughtExceptionHandler());
		/*=======啟動streams=======*/
		streams.start();
		/*=======添加jvm hook 確保streams安全退出=======*/
		Runtime.getRuntime().addShutdownHook(new Thread(() -> {
			log.info("關(guān)閉 Kafka Streams:{}...", appId);
			streams.close();
			log.info("Kafka Streams:{}已經(jīng)關(guān)閉！", appId);
		}));
	}
}

@Slf4j
public class CustomStreamsUncaughtExceptionHandler implements StreamsUncaughtExceptionHandler {
	/**
	 * Inspect the exception received in a stream thread and respond with an action.
	 *
	 */
	@Override
	public StreamThreadExceptionResponse handle(Throwable throwable) {
		log.error("Kafka Streams 線程發(fā)生未捕獲異常: {}", ExceptionUtil.stacktraceToString(throwable));
		// 選擇處理策略（以下三選一）：
		// 1. 替換線程（繼續(xù)運行）
		return StreamThreadExceptionResponse.REPLACE_THREAD;
		// 2. 關(guān)閉整個 Streams 應(yīng)用
		// return StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
		// 3. 關(guān)閉整個 JVM
		// return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
	}
}