Apache Kafka 官方文档：开始

入门Kafka 作为一个分布式的流平台，这到底意味着什么？首先来了解一下 Kafka 所使用的基本术语：主题和日志（Topic 和 Log）分布式（Distribution）Geo-Replication（异地数据同步技术）生产者（Producer）消费者（Consumer）Kafka的保证（Guarantees）Kafka作为一个消息系统Kafka 作为一个存储系统Kafka 的流处理拼在一起Kafka 的使用场景消息网站活动追踪指标日志聚合流处理事件采集提交日志Kafka 安装和启动Step 1：下载代码Step 2：启动服务Step 3：创建一个主题（topic）Step 4：发送消息Step 5：消费消息Step 6：设置多个 broker 集群Step 7：使用 Kafka Connect 来导入 / 导出 数据Step 8：使用 Kafka Stream 来处理数据

/295

入门

Kafka 作为一个分布式的流平台，这到底意味着什么？

我们认为，一个流处理平台具有三个关键能力：

发布和订阅消息（流），在这方面，它类似于一个消息队列或企业消息系统。以 容错 的方式存储消息（流）在消息流发生时处理它们。

什么是 Kafka 的优势？

它应用于 2 大类应用：

构建实时的流数据管道，可靠地获取系统和应用程序之间的数据。构建实时流的应用程序，对数据流进行转换或反应。

要了解 kafka 是如何做这些事情的，堂我们从下到上深入探讨 kafka的 能力。

首先几个概念：

kafka 作为一个集群运行在一个或多个服务器上。kafka集群存储的消息是以 topic 为类别记录的。每个消息（也叫记录 record，我们习惯叫消息）是由一个 key ，一个value和时间戳构成。

kafka 有四个核心 API：

应用程序使用 Producer API 发布消息到 1 个或多个 topic（主题）应用程序使用 Consumer API 来订阅一个或多个 topic ，并处理产生的消息。应用程序使用 Streams API 充当一个流处理器，从1 个或多个 topic 消费输入流，并生产一个输出流到1个或多个输出 topic ，有效地将输入流转换到输出流。Connect API 允许构建或运行可重复使用的 生产者或消费者，将 topic 连接到现有的应用程序或数据系统。例如，一个诶关系数据库的连接器可捕获每一个变化。

Client 和 Server 之间的通信，是通过一条简单、高性能并且和开发语言无关的 TCP 协议。并且该协议保持与老版本的兼容。kafka 提供了Java Client （客户端）。除了Java client 外，还有非常多的 其他编程语言的 Client

首先来了解一下 Kafka 所使用的基本术语：

Topic

Kafka 将消息种子（Feed）分门别类，每一类的消息称之为一个主题（Topic）。

Producer

发布消息的对象称之为主题生产者（Kafka topic producer）

Consumer

订阅消息并处理发布的消息的种子的对象称之为主题消费者（consumers）

Broker

已发布的消息保存在一组服务器中，称之为 Kafka 集群。集群中的每一个服务器都是一个代理（Broker）。消费者可以订阅一个或多个主题（topic），并从 Broker 拉数据，从而消费这些已发布的消息。

主题和日志（Topic 和 Log）

让我们更深入的了解 Kafka 中的 Topic。

Topic 是发布的消息的类别或者种子 Feed 名。对于每一个 Topic ，Kafka 集群维护这一个分区的 log，就像下图中的示例：

每一个分区都是一个顺序的、不可变的消息队列，并且可以持续的添加。分区中的消息都被分了一个序列号，称之为偏移量（offset），在每个分区中此偏移量都是唯一的。

Kafka 集群保持所有的消息，直到它们过期，无论消息是否被消费了。实际上消费者所持有的仅有的元数据就是这个偏移量，也就是消费者在这个 log 中的位置。这个偏移量由消费者控制：正常情况当消费者消费消息的时候，偏移量也线性的增加，但是实际偏移量由消费者控制，消费者可以将偏移量重置为更老的一个偏移量，重新读取消息。可以看到这种设计对消费者来说操作自如，一个消费者的操作不会影响其他消费者对此log的处理。再说说分区。 Kafka 中采用分区的设计有几个目的。一是可以处理更多的消息，不受单台服务器的限制。Topic 拥有多个分区意味着它可以不受限的处理更多的数据。第二，分区可以作为并行处理的单元，稍后会谈到这一点。

分布式（Distribution）

Log 的分区被分布到集群中的多个服务器上，每个度武器处理它分到的分区。根据配置每个分区还可以复制到其它服务器作为备份容错。每个分区有一个 leader，零或多个 follower。Leader处理此分区的所有的读写请求，而 follower 被动的复制数据。如果 leader 宕机，其他的一个 follower 会被推荐为新的 leader。一台服务器可能同时是一个分区的 leader，另一个分区的 follower。这样可以平衡负载，避免所有的请求都只让一台或者某几台服务器处理。

Geo-Replication（异地数据同步技术）

Kafka MirrorMaker 为集群提供 geo-replication 支持。借助 MirrorMaker ，消息可以跨多个数据中心或云区域进行复制。您可以在 active/passive 场景中用于备份和恢复；或者在 active/passive 方案中将数据置于更接近用户的位置，或数据本地化。

生产者（Producer）

生产者往某个 Topic 上发布消息。生产者也负责选择发布到 Topic 上的哪一个分区。最简单的方式从分区列表中轮流选择。也可以根据某种算法依照权重选择分区。开发者负责如何选择分区的算法。

消费者（Consumer）

通常来讲，消息模型可以分为两种，队列和发布订阅。队列的处理方式是一组消费者从服务器读取消息，一条消息只有其中的一个消费者来处理。在发布-订阅模型中，消息被广播给所有的消费者，接收到消息的消费者都可以处理此消息。 Kafka 为这两种模型提供了单一的消费者抽象模型：消费者组（consumer group）。消费者用一个消费者组名标记自己。一个发布在 Topic 上消息被分发给此消费者组中的一个消费者。假如所有的消费者都在一个组中，那么这就变成了 queue 模型。假如所有的消费者都在不同的组中，那么就完全变成了 发布-订阅模型。更通用的，我们可以创建一些消费者组作为逻辑上的订阅者。每个组包含数目不等的消费者，一个组内多个消费者可以用来扩展性能和容错，正如下图所示：

2 个kafka 集群托管 4 个分区（P0-P3），2个消费者组，消费组A 有2 个消费者实例，消费组 B 有 4个。

正像创痛的消息系统一样，Kafka 保证消息的顺序不变。传统的队列模型保持消息，并且保证它们的先后顺序不变。但是，金福安服务器保证了消息的顺序，消息还是异步的发送给各个消费者，消费者收到消息的先后就不能保证了。这也意味着并行消费将不能保证消息的先后循序。用过传统的消息系统的同学可定清楚，消息的顺序处理很让人头痛。如果只让一个消费者处理消息，又违背了并行处理的初衷。在这一点上 Kafka做的更好，尽管并没有完全解决上述问题。Kafka 采用了一种分而治之的策略：分区。因为 Topic 分区中消息只能由消费者组中的唯一一个消费者处理，所以消息肯定是按照先后顺序进行处理的。但是它也仅仅是保证 Topic 的一个分区顺序处理，不能保证跨分区的消息先后处理顺序。所以，如果你想要顺序的处理 Topic 的所有消息，那就只提供一个分区。

Kafka的保证（Guarantees）

生产者发送到一个特点的 Topic 的分区上，消息将会按照它们发送的顺序依次加入，也就是说，如果一个消息 M1 和 M2 使用相同的 producer 发送，M1 先发送，那么 M1将比 M2 的offset 低，并且优先的出现在日志中。消费者收到的消息也是此顺序。如果一个 Topic 配置了复制因子 （replication factor）为N，那么可以允许 N-1 服务器宕机而不丢失任何已经提交（committed）的消息

有关这些保证的更多详细信息，请参见文档的设计部分。

Kafka作为一个消息系统

Kafka 的流与传统企业消息系统相比的概念如何？

传统的消息有两种模式：队列 和 发布订阅。在队列模式中，消费者池从服务器读取消息（每个消息只被其中一个读取）；发布订阅模式：消息广播给所有的消费者。这两种模式都有优缺点，队列的优点是允许多个消费者瓜分处理数据，这样可以扩an处理。但是，队列不像多个订阅者，一旦消费者进程读取后故障了，那么消息就丢了。而发布和订阅而发布和订阅 允许你广播消息给多个消费者组（不同名）。

Kafka 的每个 Topic 都具有这两种模式。

Kafka 有比传统的消息系统更强的顺序保证。

传统的消息系统按顺序保存数据，如果多个消费者从队列消费，则服务器按存储的顺序发送消息，但是，尽管服务器按顺序发送，消息异步传递到消费者，因此消息可能乱序到达消费者。这意味着消息存在并行消费的情况，顺序就无法保证。消息系统个常常通过仅设 1 个消费者来解决这个问题，但是这意味着没用到并行处理。

Kafka 做的更好。通过并行 Topic 的 partition —— kafka 提供了顺序保证和负载均衡。每个 partition 仅由用一个消费者组中的一个消费者消费到。并确保消费者是该 partition 的唯一消费者，并并按顺序消费数据。每个 topic 有多个分区，则需要对多个消费者做负载均衡，但请注意，相同的消费者组汇总不能有比分区更多的消费者，否则多出的消费者一直处于空等待，不会受到消息。

Kafka 作为一个存储系统

所有发布消息到 消息队列 和 消费分离 的系统，实际上都充当了一个存储系统（发布的消息先存储起来）。Kafka 比别的系统的优势是它是一个非常高性能的存储系统。

写入到 Kafka 的数据将写到磁盘并复制到集群中保证容错性。并允许生产者等待应答，直到消息完全写入。

Kafka 的磁盘结构 —— 无论你服务器上有 50 KB 或 50TB，执行是相同的。

client 来控制读取数据的位置。你还可以认为 kafka 是一种专用于高性能、低延迟，提交日志存储，复制，和传播特殊用途的 分布式文件系统。

Kafka 的流处理

仅仅读，写和存储是不够的，kafka 的目标是实时的流处理。

在 kafka 中，流处理持续获取 输入 topic 的数据，进行处理加工，然后写入输出 topic。例如，一个零售 APP ，接收销售和出货的 输入流，统计数量或调整价格后输出。

可以直接使用 producer 和 consumer API 进行简单的处理。对于复杂的转换，Kafka 提供了更强大的 Streams API。可构建 聚合计算 或连接流到一起的复杂应用程序。

助于解决此类应用面临的硬性问题：处理无序的数据，代码更改的再处理，执行状态计算等。

Streams API 在 Kafka 中的核心：使用 producer 和 consumer API 作为输入，利用 Kafka 做状态存储，使用相同的组机制在 stream 处理器实例之间进行容错保障。

拼在一起

消息传递，存储和流处理的组合看似反常，但对 Kafka 作为流失处理平台的作用至关重要。

像 HDFS 这样的分布式文件系统允许存储静态文件来进行批处理。这样徐彤可以有效地存储和处理来自过去的历史数据。

传统企业的消息系统允许在你订阅之后处理未来的消息：在未来数据到达时处理它。

Kafka 结合了这两种能力，这种组合对于 kafka 作为流处理应用和 流数据管道平台是至关重要的。

批处理以及消息驱动应用程序的流处理的概念：通过组合存储和低延迟订阅，流处理应用可以用相同的方式对待过去和未来的数据。它是一个单一的应用程序，它可以处理历史的存储数据，当它处理到最后一个消息时，它进入等待未来的数据到达，而不是结束。

同样，对流数据管道（pipeline），订阅实时事件的组合使得可以将 Kafka 同于非常低延迟的管道；但是，可靠地存储数据的能力使得它可以将其用于必须保证传递的关键数据，或与仅定期加载数据或长时间维护的离线系统集成在一起。流处理可以在数据到达时转换它。

Kafka 的使用场景

下面是一些关于 Apache Kafka 流行的使用场景。这些领域的概述，可查看博客文章。

消息

kafka 更好的替换传统的消息系统，消息系统被用于各种场景（解耦数据生产者、缓存未处理的消息等），与大多数消息系统比较， Kafka 有更好的吞吐量，内置分区，副本和故障转移，这有利于处理大规模的消息。

根据我们的经验，消息往往用于较低的吞吐量，但需要低的 端到端 延迟，并需要提供强大的耐用性的保证。

在这一领域的 kafka 比得上传统的消息系统，如 ActiveMQ 或 RabbitMQ 的。

网站活动追踪

Kafka 原本的使用场景：用户的活动追踪，网站的活动（网页游戏，搜索或其他用户的操作信息）发布到不同的话题中心，这些消息可实时处理，实时监测，也可以加载到 Hadoop或离线处理数据仓库。

每个用户页面视图都会产生非常高的量。

指标

Kafka 也常常用于监测数据。分布式应用程序生成的统计数据集中聚合。

日志聚合

许多人使用 Kafka 作为日志聚合解决方案的替代品。日志聚合通常从服务器中收集物理日志文件，并将它们放在中央位置（可能是文件服务器或 HDFS）进行处理。Kafka 抽象出文件的细节，并将入职或事件数据更清晰地抽象为消息流。这允许更低延迟的处理并更容易支持多个数据源和分布式数据消费。

流处理

kafka 中消息吹一般包含多个阶段，其中原始输入数据是从 kafka 主题消费的，然后汇总，丰富，或者以其他的方式处理转化为新主题，例如，一个推荐新闻文章，文章内容可能从 “articles” 主题获取；然后进一步处理内容，得到一个处理后的新内容，最后推荐给用户。这种处理是基于单个主题的实时数据流。从 0.10.0.0 开始，轻量，但功能强大的流处理，就可以这样进行数据处理了。

除了 Kafka Streams，还有 Apache Storm 和 Apache Samza 可选择。

事件采集

事件采集是一种应用程序的设计风格，其中状态的变化根据时间的顺序记录下来， Kafka 支持这种非常大的存储日志数据的场景。

提交日志

Kafka 可以作为一种分布式的外部日志，可帮助节点之间复制数据，并作为失败的节点来恢复数据重新同步，kafka的日志压缩

功能很好的支持这种用法，这种用法类似于 Apache BookKeeper 项目。

Kafka 安装和启动

kafka 的背景知识已经讲了很多了，让我们现在开始实践吧，假设你现在没有 Kafka 和 ZooKeeper 环境。

Step 1：下载代码

下载 1.1.0 版本并且解压它。

> tar -xzf kafka_2.12-2.2.0.tgz> cd kafka_2.12-2.2.0

Step 2：启动服务

运行 kafka 需要使用 Zookeeper ，所以你需要先启动 Zookeeper，如果你没有 Zookeeper，你可以使用 Kafka 自带打包和配置好的 Zookeeper。

> bin/zookeeper-server-start.sh config/zookeeper.properties[-04-22 15:01:37,495] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)...

现在启动 Kafka 服务

> bin/kafka-server-start.sh config/server.properties &[-04-22 15:01:47,028] INFO Verifying properties (kafka.utils.VerifiableProperties)[-04-22 15:01:47,051] INFO Property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.VerifiableProperties)...

Step 3：创建一个主题（topic）

创建一个名为 “test” 的 Topic，只有一个分区和一个备份：

> bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

创建好之后，可以通过运行以下命令，查看已创建的 topic 信息：

> bin/kafka-topics.sh --list --bootstrap-server localhost:9092test

或者，除了手工创建 topic 外，你也可以配置你的 broker，当发布一个不存在的 topic 时自动创建 topic 。

Step 4：发送消息

Kafka 提供了一个命令行的工具，可以从输入文件或者命令行中读取消息并发送给 Kafka 集群。每一行是一条消息。运行 producer（生产者），然后在控制台输入几条消息到服务器。

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testThis is a messageThis is another message

Step 5：消费消息

Kafka 也提供了一个消费消息的命令行工具，将存储的信息输出出来。

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginningThis is a messageThis is another message

如果你有2台不同的终端上运行上述命令 ，那么当你的运行生产者时，消费者就能消费到生产者发送的消息。

Step 6：设置多个 broker 集群

到目前，我们只是单一的运行一个 broker，没什么意思，对于 Kafka，一个broker 仅仅只是一个集群的大小，所有让我们多设几个 broker。

首先为每个 broker 创建一个配置文件：

> cp config/server.properties config/server-1.properties > cp config/server.properties config/server-2.properties

现在编辑这些新建的文件，设置以下属性：

config/server-1.properties: broker.id=1 listeners=PLAINTEXT://:9093 log.dir=/tmp/kafka-logs-1config/server-2.properties: broker.id=2 listeners=PLAINTEXT://:9094 log.dir=/tmp/kafka-logs-2

broker.id 是集群中每个节点的唯一且永久的名称，我们修改端口和日志目录是因为我们现在在同一台机器上运行，我们要防止 broker 在同一端口上注册和覆盖对方的数据。

我们运行了 zookeeper 和刚才的一个 kafka 节点，所有我们只需要在启动 2 个新的 kafka 节点。

> bin/kafka-server-start.sh config/server-1.properties &... > bin/kafka-server-start.sh config/server-2.properties &...

现在，我们创建一个新 topic ，把备份设置为 3

> bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 1 --topic my-replicated-topic

好了，现在我们已经有了一个集群了，我们怎么知道每个集群在做什么呢？运行命令 “describe topics”

> bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic my-replicated-topicTopic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:Topic: my-replicated-topic Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0

输出解释：第一行是所有分区的摘要，其次，每一行提供一个分区信息，因为我们只有一个分区，所以只有一行。

“leader”：该节点负责该分区的所有的读和写，每个节点的 leader 都是随机选择的。“replicas”：备份的节点列表，无论该节点是否是 leader 或者目前是否还活着，只是显示。“isr”：“同步备份”的节点列表，也就是活着的节点并且正在同步 leader 。

我们运行这个命令，看看一开始我们创建的那个节点：

> bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic testTopic:test PartitionCount:1 ReplicationFactor:1 Configs:Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0

这并不奇怪，刚才创建的主题没有 Replicas，并且在服务器 “0” 上，我们创建它的时候，集群中只有一个服务器，所以，是“0”。

让我们来发布一些信息在新的 topic 上：

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic...my test message 1my test message 2^C

现在，消费这些消息

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic...> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic...my test message 1my test message 2^C

我们要测试集群的容错，kill 掉 leader， broker1 作为当前的 leader ，也就是 kill 掉 broker1.

> ps | grep server-1.properties7564 ttys002 0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.6/Home/bin/java... > kill -9 7564

在 Windows 上使用：

> wmic process where "caption = 'java.exe' and commandline like '%server-1.properties%'" get processidProcessId6016> taskkill /pid 6016 /f

备份节点之一称为新的 leader ，而 broker1已经不在同步备份集合里了。

> bin/kafka-topics.sh --describe --broker-list localhost:9092 --topic my-replicated-topicTopic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:Topic: my-replicated-topic Partition: 0 Leader: 2 Replicas: 1,2,0 Isr: 2,0

但是，消息仍然没丢：

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic...my test message 1my test message 2^C

Step 7：使用 Kafka Connect 来导入 / 导出 数据

从控制台写入和写回数据时一个方便的开始，但你可能想要从其他来源导入或导出数据到其他系统。对于大多数系统，可以使用 kafka connect，而不需要编写自定义集成代码。

Kafka Connect 是 导入和导出 数据的一个工具，它是一个可扩展的工具，运行连接器，实现与自定义的逻辑的外部系统的交互。在这个快速入门里，我们将看到如何运行 Kafka Connect用简单的连接器从文件导入数据到 kafka主题，再从 Kafka 主题导出数据到文件。

首先，我们首先创建一些 “种子” 数据用来测试，（ps：种子的意思就是造一些消息）：

echo -e "foo\nbar" > test.txt

windows 上：

> echo foo> test.txt> echo bar>> test.txt

接下来，我们开始 2 个连接器运行在独立的模式，这意味着他们运行在一个单一的，本地的，专用的进程。我们提供 3个 配置文件作为参数。首先是 Kafka Conntect 处理的配置，包含常见的配置，例如要连接的 Kafka broker和 数据的序列化格式。其余的配置文件指定了要创建的连接器。包括连接器唯一名称，和要实例化的连接器类。以及连接器所需的任何其他配置。

> bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

kafka 附带了这些示例的配置文件，并且使用了刚才我们搭建的本地集群配置并创建了2个连接器：第一个是源连接器，从输入文件中读取并发布到 Kafka 主题汇总，第二个是接收连接器，从 kafka 主题读取消息输出到外部文件。

在启动过程中，你会看到一些日志消息，包括一些连接器实例化的说明。一旦 kafka connect 进程已经开始，导入连接器应该读取从

test.txt

和写入到 topic

connect-test

导出连接器从主题

connect-test

读取消息写入到文件

test.slink.txt

我们可以通过验证输出文件的内容来验证数据已经全部导出：

more test.sink.txtfoobar

注意，导入的数据也已经在 Kafka 主题

contect-test

里，所以我们可以使用该命令查看这个主题：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning{"schema":{"type":"string","optional":false},"payload":"foo"}{"schema":{"type":"string","optional":false},"payload":"bar"}...

连接器继续处理数据，因此我们可以添加数据到文件并通过管道移动：

echo "Another line" >> test.txt

你应该会看到出现在消费者控制台输出一行信息并导出到文件。

Step 8：使用 Kafka Stream 来处理数据

Kafka Stream 是 kafka 的客户端库，用于实时流处理和分析存储在 kafka broker 的数据，这个快速入门示例将演示如何运行一个流应用程序。一个 WordCountDemo 的例子（为了方便阅读，使用的是 java8 lambda 表达式）

KTable wordCounts = textLines// Split each text line, by whitespace, into words..flatMapValues(value -> Arrays.asList(value.toLowerCase().split("W+")))// Ensure the words are available as record keys for the next aggregate operation..map((key, value) -> new KeyValue<>(value, value))// Count the occurrences of each word (record key) and store the results into a table named "Counts"..countByKey("Counts")

它实现了 wordcount 算法，从输入的文本计算出一个词出现的次数。然而，不像其他的 word count 的例子，你可能会看到，在有限的数据之前，执行的演示应用程序的行为略有不同，因为它的目的是在一个无限的操作，数据流。类似的有界变量，它是哟中动态算法，跟踪和更新的单词计数。然而，由于它必须假设潜在的无界输入数据，它会定期地输出其当前状态和结果，同时继续处理更多的数据，因为它不知道什么时候它处理过的 “所有” 的输入数据。

现在准备输入数据到 kafka 的 topic 汇总，随后 kafka stream 应用处理这个 topic 的数据。

> echo -e "all streams lead to kafka\nhello kafka streams\njoin kafka summit" > file-input.txt

接下来，使用控制台的 producer 将输入的数据发送到 指定的 topic （streams-file-input）中，（在实践中，stream 数据可能会持续流入，其中 kafka 的应用将启动并运行）

> bin/kafka-topics.sh --create \--bootstrap-server localhost:9092 \--replication-factor 1 \--partitions 1 \--topic streams-file-input

> cat /tmp/file-input.txt | ./bin/kafka-console-producer --broker-list localhost:9092 --topic streams-file-input

现在，我们运行 WordCount 处理输入的数据：

> ./bin/kafka-run-class org.apache.kafka.streams.examples.wordcount.WordCountDemo

不会有任何的 STDOUT 输出，除了日志，结果不断地写回另一个 topic （streams-wordcount-output），demo 运行几秒，然后，不像典型的流处理应用程序，自动终止。

现在我们检查 WordCountDemo 应用，从输出的 topic 读取。

> ./bin/kafka-console-consumer --zookeeper localhost:2181 --topic streams-wordcount-output --from-beginning --formatter kafka.tools.DefaultMessageFormatter --property print.key=true --property print.key=true --property key.deserializer=org.mon.serialization.StringDeserializer --property value.deserializer=org.mon.serialization.LongDeserializer

输出数据打印到控制台（你可以使用 Ctrl+C 停止）

all1streams 1lead 1to1kafka 1hello 1kafka 2streams 2join 1kafka 3summit 1^C

第一列 是 message 的 key ，第二列是 message 的value，要注意，输出的实际是一个连续的更新流，其中每条数据（即：原始输出的每行）是一个单词的最新的 count，又叫记录键 “kafka”。对于同一个 key 有多个记录，每个记录之后是前一个的更新。