Kafka详解与总结(二)

心已赠人 2022-01-10 04:47 217阅读 0赞

## **Kafka Stream** ##

Kafka Streams是一个客户端库，用于构建任务关键型实时应用程序和微服务，其中输入和输出数据存储在Kafka集群中。Kafka Streams结合了在客户端编写和部署标准Java和Scala应用程序的简单性以及Kafka服务器端集群技术的优势，使这些应用程序具有高度可扩展性，弹性，容错性，分布式等等。

以下是[WordCountDemo][]示例代码的要点（为了方便阅读，使用的是java8 lambda表达式）。

步骤：

1.启动zk和kafka

> bin/zookeeper-server-start.sh config/zookeeper.properties

> bin/kafka-server-start.sh config/server.properties

1.  准备输入主题并启动生产者

创建名为**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

1.  启动Wordcount应用程序

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

演示应用程序将从输入主题**stream-plaintext-input**读取，对每个读取消息执行WordCount算法的计算，并将其当前结果连续写入输出主题**streams-wordcount-output**。

1.  处理数据

开启一个生产者终端：

> 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格式的消息值，表示单词的最新计数。

转载于:https://www.cnblogs.com/htkj/p/10941438.html

[WordCountDemo]: https://github.com/apache/kafka/blob/2.1/streams/examples/src/main/java/org/apache/kafka/streams/examples/wordcount/WordCountDemo.java

发表评论取消回复

表情：

评论列表（有 0 条评论，217人围观）

还没有评论，来说两句吧...

相关阅读

相关 kafka详解二：如何配置kafka集群

[http://www.360doc.com/content/17/1020/10/16915\_696580124.shtml][http_www.360doc.com_co

缺乏、安全感/ 2022年09月10日 01:26/ 0 赞/ 211 阅读

相关 Kafka（二）： Kafka 集群部署与使用

一、Kafka 集群部署 Kafk

秒速五厘米/ 2022年06月15日 02:13/ 0 赞/ 218 阅读

相关 Kafka（二）：Kafka安装与使用

一、安装Kafka 官方快速安装文档：[http://kafka.apache.org/quickstart][http_kafka.apache.org_quickst

迷南。/ 2022年05月27日 08:49/ 0 赞/ 295 阅读

相关 Kafka Producer详解(二)

文章目录一、Producer介绍同步和异步的方式调用send() 二、Producer工作原理介绍

骑猪看日落/ 2022年05月18日 07:58/ 0 赞/ 245 阅读

相关 Kafka总结（九）：KafKa 与Spark整合应用

[Kafka总结（一）：Kafka概述][Kafka_Kafka] [Kafka总结（二）：Kafka核心组件][Kafka_Kafka 1] [Kafka总结（三）：Ka

冷不防/ 2022年05月18日 04:41/ 0 赞/ 459 阅读

相关 Kafka总结（八）：KafKa与ELK整合应用

[Kafka总结（一）：Kafka概述][Kafka_Kafka] [Kafka总结（二）：Kafka核心组件][Kafka_Kafka 1] [Kafka总结（三）：Ka

╰半夏微凉°/ 2022年05月18日 04:41/ 0 赞/ 248 阅读

相关 Kafka总结（六）：Kafka Stream详解

[Kafka总结（一）：Kafka概述][Kafka_Kafka] [Kafka总结（二）：Kafka核心组件][Kafka_Kafka 1] [Kafka总结（三）：Ka

古城微笑少年丶/ 2022年05月18日 04:41/ 0 赞/ 846 阅读

相关 Kafka总结（五）：API编程详解

[Kafka总结（一）：Kafka概述][Kafka_Kafka] [Kafka总结（二）：Kafka核心组件][Kafka_Kafka 1] [Kafka总结（三）：Ka

野性酷女/ 2022年05月18日 04:41/ 0 赞/ 387 阅读

相关 Kafka总结（二）：核心组件

[Kafka总结（一）：Kafka概述][Kafka_Kafka] [Kafka总结（二）：Kafka核心组件][Kafka_Kafka 1] [Kafka总结（三）：Ka

青旅半醒/ 2022年05月18日 04:38/ 0 赞/ 238 阅读

相关 Kafka详解与总结(二)

Kafka Stream Kafka Streams是一个客户端库，用于构建任务关键型实时应用程序和微服务，其中输入和输出数据存储在Kafka集群中。Kafka Stre

心已赠人/ 2022年01月10日 04:47/ 0 赞/ 218 阅读