/* 用于實現將一個topic的數據讀取后從轉換成大寫，再將結果寫入另一個topic*/ public class StreamAPI {public static void main(String[] args) {Properties props = new Properties();//設置程序的唯一標識props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");//設置kafka集群props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "node01:9092,node02:9092,node03:9092");//設置序列化與反序列化props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());//實例一個計算邏輯StreamsBuilder streamsBuilder = new StreamsBuilder();//設置計算邏輯 stream：讀取 to：寫入streamsBuilder.stream("18BD34").mapValues(line->line.toString().toUpperCase()).to("18BD34-1");//構建Topology對象（拓撲，流程）final Topology topology = streamsBuilder.build();//實例 kafka流KafkaStreams streams = new KafkaStreams(topology, props);//啟動流計算streams.start();}}

總結

以上是生活随笔為你收集整理的Kafka中steamAPi操作的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。