update 优化 完善kafka-stream案例
parent
3e57a42f39
commit
0fe9c4f17d
@ -0,0 +1,27 @@
|
|||||||
|
package org.dromara.stream.config;
|
||||||
|
|
||||||
|
import org.apache.kafka.streams.StreamsBuilder;
|
||||||
|
import org.apache.kafka.streams.kstream.KStream;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* kafka stream 配置
|
||||||
|
*
|
||||||
|
* @author LionLi
|
||||||
|
*/
|
||||||
|
@Configuration
|
||||||
|
public class KafkaStreamsConfig {
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public KStream<String, String> demoStream(StreamsBuilder builder) {
|
||||||
|
// 输入主题
|
||||||
|
KStream<String, String> source = builder.stream("input-topic");
|
||||||
|
// 转换逻辑:这里只是简单地将消息转换为大写
|
||||||
|
KStream<String, String> processed = source.mapValues(value -> value.toUpperCase());
|
||||||
|
// 输出到另一个主题
|
||||||
|
processed.to("output-topic");
|
||||||
|
return source;
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
Loading…
Reference in New Issue