Kafka 1.0.1案例详解之Kafka Streams
首先，要研究一样新东西，我们需要知道它是做什么的：
Kafka Streams is a client library for processing and analyzing data stored in Kafka. It builds upon important stream processing concepts such as properly distinguishing between event time and processing time, windowing support, and simple yet efficient management and real-time querying of application state.
大家仔细阅读上面一段话可以知道，Kafka Streams是一个用来处理Kafka消息的库，它包含了如下几个优势：
通过与现有的Java应用整合，我们可以设计出简单的、轻量级的客户端类库
只需要基于Kafka自身的消息系统，不需要额外的第三方系统，就可以很容易地实现水平扩展
通过可容错的状态管理，实现高效的窗口操作和聚合
支持 exactly-once语义
既支持基于时间窗口的操作，也支持每次单条数据的处理
既支持低阶的流处理接口，也支持高阶的流处理DSL（领域专用语言）
Kafka Streams处理剖析图
案例剖析
说了这么多理论知识，实际上用起来很简单，接下来我们通过一个简单的例子来熟悉这个新特性。
添加依赖
kafka-streams是一个单独的依赖包，并不存在于kafka-client中
org.apache.kafka
kafka-streams
1.0.1
属性配置
添加属性配置，application id相当于group id，bootstrap servers配置kafka的brokers地址，并配置key与value的序列化、反序列化实现类。这两个类均实现了
Properties props =newProperties();
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");
finalTopology topology = builder.build();
finalKafkaStreams streams =newKafkaStreams(topology,props);
退出机制
最后添加退出时的处理逻辑
// attach shutdown handler to catch control-c
Runtime.getRuntime().addShutdownHook(newThread("streams-shutdown-hook") {
@Override
public voidrun() {
streams.close();
latch.countDown();
}
});
我们可以在github中查看完整的程序代码：
https://github.com/lubinsu/new-kafka
- 发表于:
- 原文链接：https://kuaibao.qq.com/s/20180606G1DJVJ00?refer=cp_1026
- 腾讯「腾讯云开发者社区」是腾讯内容开放平台帐号（企鹅号）传播渠道之一，根据《腾讯内容开放平台服务协议》转载发布内容。
- 如有侵权，请联系 cloudcommunity@tencent.com 删除。
相关快讯
扫码
添加站长 进交流群
领取专属 10元无门槛券
私享最新 技术干货
腾讯云开发者
