Kafka Stream
Kafka Streams是一个客户端库，用于构建任务关键型实时应用程序和微服务，其中输入和输出数据存储在Kafka集群中。Kafka Streams结合了在客户端编写和部署标准Java和Scala应用程序的简单性以及Kafka服务器端集群技术的优势，使这些应用程序具有高度可扩展性，弹性，容错性，分布式等等。
以下是示例代码的要点（为了方便阅读，使用的是java8 lambda表达式）。
步骤：
1.启动zk和kafka
> bin/zookeeper-server-start.sh config/zookeeper.properties
> bin/kafka-server-start.sh config/server.properties
- 准备输入主题并启动生产者
创建名为streams-plaintext-input的输入主题和名为streams-wordcount-output的输出主题：
> bin/kafka-topics.sh --create \
--zookeeper localhost:2181 \
--replication-factor 1 \
--partitions 1 \
--topic streams-plaintext-input
Created topic "streams-plaintext-input".
注意：我们创建输出主题并启用压缩，因为输出流是更改日志流
> bin/kafka-topics.sh --create \
--zookeeper localhost:2181 \
--replication-factor 1 \
--partitions 1 \
--topic streams-wordcount-output \
--config cleanup.policy=compact
Created topic "streams-wordcount-output".
使用相同的kafka-topics工具描述创建的主题：
> bin/kafka-topics.sh --zookeeper localhost:2181 --describe
- 启动Wordcount应用程序
> bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo
演示应用程序将从输入主题stream-plaintext-input读取，对每个读取消息执行WordCount算法的计算，并将其当前结果连续写入输出主题streams-wordcount-output。
- 处理数据
开启一个生产者终端：
> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic
streams-plaintext-input
all streams lead to kafka
开启一个消费者终端：
> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic streams-wordcount-output \
--from-beginning \
--formatter kafka.tools.DefaultMessageFormatter \
--property print.key=true \
--property print.value=true \
--property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
--property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer
all 1
streams 1
lead 1
to 1
kafka 1
这里，第一列是java.lang.String格式的Kafka消息键，表示正在计数的单词，第二列是java.lang.Long格式的消息值，表示单词的最新计数。