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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。