java操作 kafka streams 入门实例（sum+wordcount）详细流程

在idea上创建一个maven工程

然后在poml.xml文件里增加kafka-streams的依赖包:

    org.apache.kafka
    kafka-streams
    2.2.2

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;

import java.util.Properties;

public class SumStreamDemo {
    public static void main(String[] args) {
        Properties prop = new Properties();
        prop.put(StreamsConfig.APPLICATION_ID_CONFIG,"wordcount");
        prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.26.128:9092");
        prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");

        //earliest latest none
        prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
        prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());

        //如何构建流结构拓扑
        StreamsBuilder builder = new StreamsBuilder();
        //不断从mystreamin上获取新数据，并且追加到流上的一个抽象对象
        KStream source = builder.stream("mystreamin");//[4,5,6]
        //java lamda 表达式
        //KTable是数据集合的抽象对象
        KTable sum1 = source.map((key, value) -> new KeyValue("sum", value.toString()))
                .groupByKey() //[sum,(4,5,6)]
                .reduce((x, y) -> {
                    Integer sum = Integer.valueOf(x) + Integer.valueOf(y);
                    System.out.println(x + " + " + y + " = " + sum);
                    return sum.toString();
                });
        sum1.toStream().to("mystreamout");

        Topology topology = builder.build();
        KafkaStreams kafkaStreams = new KafkaStreams(topology, prop);
        kafkaStreams.start();
    }
}

wordcount

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class WordCountDemo {
    public static void main(String[] args) {
        Properties prop = new Properties();
        prop.put(StreamsConfig.APPLICATION_ID_CONFIG,"wordcount2");
        prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.26.128:9092");
        prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");

        //earliest latest none
        prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
        prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());

        //如何构建流结构拓扑
        StreamsBuilder builder = new StreamsBuilder();
        //不断从mystreamin上获取新数据，并且追加到流上的一个抽象对象
        KStream