KStream<String, String> stream = builder.stream("words-stream");
KTable<String, String> table = builder.table("words-table", "words-store");

public class WordCountProcessor implements Processor<String, String> {
    private ProcessorContext context;
    private KeyValueStore<String, Integer> kvStore;
    @SuppressWarnings("unchecked")
    @Override
    //init()方法：可以获取ProcessorContext实例，用来维护当前上下文；通过上下文ProcessorContext得到状态仓库实例以及使用上下文用于基于时间推移周期性的执行；
    public void init(ProcessorContext context) {
        this.context = context;
        this.context.schedule(1000);
        this.kvStore = (KeyValueStore<String, Integer>) context.getStateStore("Counts");
    }
    //process方法：用于对收到的数据集执行对状态仓库的操作；
    @Override
    public void process(String key, String value) {
        Stream.of(value.toLowerCase().split(" ")).forEach((String word) -> {
            Optional<Integer> counts = Optional.ofNullable(kvStore.get(word));
            int count = counts.map(wordcount -> wordcount + 1).orElse(1);
            kvStore.put(word, count);
        });
    }
    @Override
    public void punctuate(long timestamp) {   //punctuate():用于基于时间推移周期性执行；
        KeyValueIterator<String, Integer> iterator = this.kvStore.all();
        iterator.forEachRemaining(entry -> {
            context.forward(entry.key, entry.value);
            this.kvStore.delete(entry.key);
        });
        context.commit();
    }
    @Override   //close:关闭相应资源操作
    public void close() {
        this.kvStore.close();
    }
}

KStream<K, V> through(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<K, V> partitioner, String topic)

StreamsBuilder builder = new StreamsBuilder();
//StreamsBuilder.table(final String topic)创建KTable实例的同时，内部会创建一个StateStore来跟踪流的状态，但它不可用于交互式查询。
// 创建KTable实例
KTable<String, StockTickerData> stockTickerTable = builder.table("stock-ticker-table");
// 创建KStream实例
KStream<String, StockTickerData> stockTickerStream = builder.stream("stock-ticker-stream");
// 打印结果到控制台
stockTickerTable.toStream().print(Printed.<String, StockTickerData>toSysOut().withLabel("Stocks-KTable"));
stockTickerStream.print(Printed.<String, StockTickerData>toSysOut().withLabel("Stocks-KStream"));

//application id相当于group id，bootstrap servers配置kafka的brokers地址，并配置key与value的序列化、反序列化实现类。
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-pipe");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

//最后通过StreamsBuilder来创建KStream，进行数据处理转换后输出到一个新的topic或者其他外部存储器中。
builder.stream("streams-plaintext-input").to("streams-pipe-output");
final Topology topology = builder.build();
final KafkaStreams streams = new KafkaStreams(topology, props);

// attach shutdown handler to catch control-c
Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") { @Override public void run() { streams.close(); latch.countDown(); }
});

package cc.gmem.study.kafka;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class NameCountApplication {

    private static final Logger LOGGER = LogManager.getLogger( NameCountApplication.class );

    public static void main( final String[] args ) throws Exception {
        Properties config = new Properties();
        // 应用的标识符，不同的实例依据此标识符相互发现
        config.put( StreamsConfig.APPLICATION_ID_CONFIG, "names-counter-application" );
        // 启动时使用的Kafka服务器
        config.put( StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1.gmem.cc:9092" );
        // 键值串行化类
        config.put( StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass() );
        config.put( StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass() );
        // High Level DSL for building topologies
        StreamsBuilder builder = new StreamsBuilder();
        // KStream是记录（KeyValue）流的抽象
        KStream<String, String> nameBatches = builder.stream( "names" );
        KTable<String, Long> nameCounts = nameBatches
            // 一到多映射，分割字符串
            .flatMapValues( nameBatch -> Arrays.asList( nameBatch.split( "\\W+" ) ) )
            // 根据人名分组
            .groupBy( ( key, name ) -> name )
            // 进行聚合，结果存放到StateStore中
            .count( Materialized.as( "names-count-store" ) );
        // 输出到目标
        nameCounts.toStream().to( "names-count", Produced.with( Serdes.String(), Serdes.Long() ) );
        // 构建流处理程序并启动
        KafkaStreams streams = new KafkaStreams( builder.build(), config );
        LOGGER.trace( "Prepare to start stream processing." );
        streams.start();

        TimeUnit.DAYS.sleep( 1 );  // 阻塞主线程
    }
}

Topology topology = new Topology();
// 指定拓扑的输入，也就是Kafka主题
topology.addSource( "SOURCE", "src-topic" )
        // 添加一个处理器PROCESS1，其上游为拓扑输入（通过名称引用）
        .addProcessor( "PROCESS1", () -> new Processor1(), "SOURCE" )
        // 添加另一个处理器PROCESS2，以PROCESS1为上游
        .addProcessor( "PROCESS2", () -> new Processor2(), "PROCESS1" )
        // 添加另一个处理器PROCESS3，仍然以PROCESS1为上游，注意拓扑分叉了
        .addProcessor( "PROCESS3", () -> new Processor3(), "PROCESS1" )
        // 添加一个输出处理器，输出到sink-topic1，以PROCESS1为上游
        .addSink( "SINK1", "sink-topic1", "PROCESS1" )
        // 添加一个输出处理器，输出到sink-topic2，以PROCESS2为上游
        .addSink( "SINK2", "sink-topic2", "PROCESS2" )
        // 添加一个输出处理器，输出到sink-topic3，以PROCESS3为上游
        .addSink( "SINK3", "sink-topic3", "PROCESS3" );

package cc.gmem.study.kafka.streams.low;
 
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
 
public class NameCounterProcessor implements Processor<String, String> {
    private ProcessorContext context;
    private KeyValueStore<String, Long> kvStore;
    @Override
    public void init( ProcessorContext context ) {
        // 保存引用，类似于Storm的TopologyContext
        this.context = context;
        // 从上下文中取回一个状态存储
        this.kvStore = (KeyValueStore<String, Long>) context.getStateStore( "NameCounts" );
        // 以墙上时间为准，每秒执行Punctuator逻辑
        this.context.schedule( 1000, PunctuationType.WALL_CLOCK_TIME, timestamp -> {
            NameCounterProcessor.this.punctuate( timestamp );
        } );
    }
    /**
     * 接收一个记录（人名列表）并处理
     *
     * @param dummy 记录的键，无用
     * @param line  记录的值
     */
    @Override
    public void process( String dummy, String line ) {
        String[] names = line.toLowerCase().split( " " );
        // 在键值存储中更新人名计数
        for ( String name : names ) {
            Long oldCount = this.kvStore.get( name );
            if ( oldCount == null ) {
                this.kvStore.put( name, 1L );
            } else {
                this.kvStore.put( name, oldCount + 1L );
            }
        }
    }
    @Override
    public void punctuate( long timestamp ) {
        // 获得键值存储的迭代器
        KeyValueIterator<String, Long> iter = this.kvStore.all();
        while ( iter.hasNext() ) {
            KeyValue<String, Long> entry = iter.next();
            // 转发记录给下游处理器
            context.forward( entry.key, entry.value.toString() );
        }
        /**
         * 调用者必须要负责关闭状态存储上的迭代器
         * 否则可能（取决于底层状态存储的实现）导致内存、文件句柄的泄漏
         */
        iter.close();
        // 请求提交当前流状态（消费进度）
        context.commit();
    }
    @Override
    public void close() {
        // 在此关闭所有持有的资源，但是状态存储不需要关闭，由Kafka Stream自己维护其生命周期
    }
}

Topology topology = new Topology();
topology.addSource("Source", "source-topic")
    .addProcessor("Process", () -> new WordCountProcessor(), "Source")
    // 为处理器Process提供一个状态存储 
    .addStateStore(countStoreSupplier, "Process");

import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.streams.processor.StateRestoreListener;
// 监听器会被所有org.apache.kafka.streams.processor.internals.StreamThread实例共享，并必须线程安全
public class ConsoleGlobalRestoreListerner implements StateRestoreListener {
    // 在恢复过程开始时回调
    public void onRestoreStart(
            final TopicPartition topicPartition,  // 主题分区
            final String storeName,               // 状态存储名称
            final long startingOffset,            // 需要恢复的起点
            final long endingOffset               // 需要恢复的终点
    ) {}
 
    // 在恢复一批次数据后回调
    public void onBatchRestored( final TopicPartition topicPartition,
            final String storeName,
            final long batchEndOffset,
            final long numRestored 
    ) {}
 
    // 恢复完成后回调
    public void onRestoreEnd( final TopicPartition topicPartition,
            final String storeName,
            final long totalRestored ) {}
}

// 启用：StateStoreBuilder#withLoggingEnabled(Map<String, String>);
// 禁用：StateStoreBuilder#withLoggingDisabled();
KeyValueBytesStoreSupplier countStoreSupplier = Stores.inMemoryKeyValueStore("Counts");
StateStoreBuilder builder = Stores
    .keyValueStoreBuilder(countStoreSupplier,Serdes.String(),Serdes.Long())
    .withLoggingDisabled();

//通过读取Kafka主题，即可为Kafka Streams提供输入流。首先你需要实例化一个StreamsBuilder：
StreamsBuilder builder = new StreamsBuilder();
//创建 KStream 流
KStream<String, Long> nameCounts = builder.stream( 
    "names-counts-input-topic",  // 输入主题名称
    Consumed.with(Serdes.String(), Serdes.Long()) // 指定键值的串行化器
);

//创建 KTable
KTable<String, Long> nameCounts = builder.table(
    Serdes.String(), /* 键串行化器 */
    Serdes.Long(),   /* 值串行化器 */
    "name-counts-input-topic", /* 输入主题 */
    "name-counts-partitioned-store" /* 表名 */);

//创建GlobalKTable
GlobalKTable<String, Long> nameCounts = builder.globalTable(
    Serdes.String(), /* 键串行化器 */
    Serdes.Long(),   /* 值串行化器 */
    "name-counts-input-topic", /* 输入主题 */
    "name-counts-global-store" /* 表名 */);

KStream<String, Long> stream = ...;
KStream<String, Long>[] branches = stream.branch(
        (key, value) -> key.startsWith("A"), /* 以A开头的键  */
        (key, value) -> key.startsWith("B"), /* 以B开头的键 */
        (key, value) -> true                 /* 所有其它的记录均发往此流  */
);

// 仅保留正数值
stream.filter((key, value) -> value > 0);
// 针对一个KTable进行过滤，结果物化到一个StageStore中
Materialized m = Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("filtered")
table.filter((key, value) -> value != 0, m);

KStream<String, Integer> transformed = stream.flatMap(
    (key, value) -> {
        // 键值对的列表
        List<KeyValue<String, Integer>> result = new LinkedList<>();
        result.add(KeyValue.pair(value.toUpperCase(), 1000));
        result.add(KeyValue.pair(value.toLowerCase(), 9000));
        return result;
    }
);

KGroupedStream<byte[], String> groupedStream = stream.groupByKey(
    // 如果键值的类型不匹配配置的默认串行化器，则需要明确指定：
    Serialized.with(
         Serdes.ByteArray(),
         Serdes.String())
);

KGroupedStream<String, String> groupedStream = stream.groupBy(
    (key, value) -> value,  // 产生键值对value:value并依此分组
    Serialize.with(
         Serdes.String(), /* 键的类型发生改变 */
         Serdes.String())  /* value */
); 
KGroupedTable<String, Integer> groupedTable = table.groupBy(
    // 产生键值对  value:length(value)，并依此分组
    (key, value) -> KeyValue.pair(value, value.length()),
    Serialized.with(
        Serdes.String(), /* 键的类发生改变 */
        Serdes.Integer()) /* 值的类型发生改变  */
);

KStream<byte[], String> stream = ...;
KStream<String, Integer> transformed = stream.map(
    (key, value) -> KeyValue.pair(value.toLowerCase(), value.length()));

KStream<String, String> rekeyed = stream.selectKey((key, value) -> value.split(" ")[0])

KGroupedStream<Bytes, String> groupedStream = null;
KGroupedTable<Bytes, String> groupedTable = null;
// 聚合一个分组流，值类型从字符串变为整数
KTable<Bytes, Long> aggregatedStream = groupedStream.aggregate(
    () -> 0L, /* 初始化器 */
    ( aggKey, newValue, aggValue ) -> aggValue + newValue.length(), /* 累加器 */
    Serdes.Long(), /* 值的串行化器 */
    "aggregated-stream-store" /* 状态存储的名称 */ );

KTable<Bytes, Long> aggregatedTable = groupedTable.aggregate(
    () -> 0L, /* 初始化器 */
    ( aggKey, newValue, aggValue ) -> aggValue + newValue.length(), /* 累加器 */
    ( aggKey, oldValue, aggValue ) -> aggValue - oldValue.length(), /* 减法器 */
    Serdes.Long(), /* 值的串行化器 */
    "aggregated-table-store" /* 状态存储的名称 */ );

KGroupedStream<String, Long> groupedStream = null;

// 基于时间的窗口化（滚动窗口）
KTable<Windowed<String>, Long> timeWindowedAggregatedStream = groupedStream
    .windowedBy( TimeWindows.of( TimeUnit.MINUTES.toMillis( 5 ) ) )
    .aggregate(
    () -> 0L, /* 初始化器 */
    ( aggKey, newValue, aggValue ) -> aggValue + newValue, /* 累加器 */
    /* 状态存储 */
    Materialized.<String, Long, WindowStore<Bytes, byte[]>>as( "time-windowed-aggregated-stream-store" )
    .withValueSerde( Serdes.Long() ) );
// 基于会话的窗口化
KTable<Windowed<String>, Long> sessionizedAggregatedStream = groupedStream
    .windowedBy( SessionWindows.with( TimeUnit.MINUTES.toMillis( 5 ) ) ) /* 窗口定义 */
    .aggregate(
    () -> 0L, /* 初始化器 */
    ( aggKey, newValue, aggValue ) -> aggValue + newValue, /* 累加器 */
    ( aggKey, leftAggValue, rightAggValue ) -> leftAggValue + rightAggValue, /* 会话合并器 */
    Materialized.<String, Long, SessionStore<Bytes, byte[]>>as( "sessionized-aggregated-stream-store" ).withValueSerde( Serdes.Long() ) );