RocketMQ Streams 1.1.0: 轻量级流处理再出发

本文作者：倪泽，Apache RocketMQ committer、RSQLDB/RocketMQ Streams Maintainer

01 背景

RocketMQ Streams是一款基于RocketMQ为基础的轻量级流计算引擎，具有资源消耗少、部署简单、功能全面的特点，目前已经在社区开源。RocketMQ Streams在阿里云内部被使用在对资源比较敏感，同时又强烈需要流计算的场景，比如在自建机房的云安全场景下。

自RocketMQ Streams开源以来，吸引了大量用户调研和试用。但是也存在一些问题，在RocketMQ Streams 1.1.0中，主要针对以下问题做出了改进和优化。

1、面向用户API不够友好，不能使用泛型，不支持自定义序列化/反序列化；

2、代码冗余，在RocketMQ Streams中存在将流处理拓扑序列化反序列化模块，RocketMQ Streams作为轻量级流处理SDK，构建好流处理节点之后应该可以直接处理数据，不存在将流处理拓扑图本地保存或者网络传输需求。

3、流处理过程不容易理解，含有大量缓存、刷新逻辑；

4、存在大量支持SQL的代码，这部分和SDK方式运行流处理任务的逻辑无关；

在RocketMQ Streams 1.1.0中，对上述问题做出了改进，期望能带来更好的使用体验。同时，重新设计了流处理拓扑构建过程、去掉冗余代码，使得代码更容易被理解。

从今天起，将推出系列文章介绍RocketMQ Streams 1.1.0版本，本次文章主要介绍RocketMQ Streams 1.1.0的API如何使用，如何利用RocketMQ Streams快速构建流处理应用。

02 典型使用示例

本地运行下列示例的步骤：

1、部署RocketMQ 5.0；

2、使用mqAdmin创建topic；

3、构建示例工程，添加依赖，启动示例。RocketMQ Streams 坐标：

<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-streams</artifactId><version>1.1.0</version>
</dependency>

4、向topic中写入相应数据，并观察结果。

更详细文档请参考：GitHub - apache/rocketmq-streams: Apache rocketmq

WordCount

public class WordCount {public static void main(String[] args) {StreamBuilder builder = new StreamBuilder("wordCount");builder.source("sourceTopic", total -> {String value = new String(total, StandardCharsets.UTF_8);return new Pair<>(null, value);}).flatMap((ValueMapperAction<String, List<String>>) value -> {String[] splits = value.toLowerCase().split("\W+");return Arrays.asList(splits);}).keyBy(value -> value).count().toRStream().print();TopologyBuilder topologyBuilder = builder.build();Properties properties = new Properties();properties.put(MixAll.NAMESRV_ADDR_PROPERTY, "127.0.0.1:9876");RocketMQStream rocketMQStream = new RocketMQStream(topologyBuilder, properties);final CountDownLatch latch = new CountDownLatch(1);Runtime.getRuntime().addShutdownHook(new Thread("wordcount-shutdown-hook") {@Overridepublic void run() {rocketMQStream.stop();latch.countDown();}});try {rocketMQStream.start();latch.await();} catch (final Throwable e) {System.exit(1);}System.exit(0);}
}

WordCount示例要点：

1、JobId wordCount唯一标识流处理任务；

2、自定义的反序列化；

3、一对多转化；

4、lambda形式从数据中指定Key；

5、支持有状态计算；

窗口聚合

public class WindowCount {public static void main(String[] args) {StreamBuilder builder = new StreamBuilder("windowCountUser");AggregateAction<String, User, Num> aggregateAction = (key, value, accumulator) -> new Num(value.getName(), 100);builder.source("user", source -> {User user1 = JSON.parseObject(source, User.class);return new Pair<>(null, user1);}).selectTimestamp(User::getTimestamp).filter(value -> value.getAge() > 0).keyBy(value -> "key").window(WindowBuilder.tumblingWindow(Time.seconds(15))).aggregate(aggregateAction).toRStream().print();TopologyBuilder topologyBuilder = builder.build();Properties properties = new Properties();properties.putIfAbsent(MixAll.NAMESRV_ADDR_PROPERTY, "127.0.0.1:9876");properties.put(Constant.TIME_TYPE, TimeType.EVENT_TIME);properties.put(Constant.ALLOW_LATENESS_MILLISECOND, 2000);RocketMQStream rocketMQStream = new RocketMQStream(topologyBuilder, properties);rocketMQStream.start();}
}

窗口聚合示例要点：

1、支持指定时间字段；

2、支持滑动、滚动、会话多种类型window；

3、支持自定义UDAF类型聚合；

4、支持自定义时间类型和数据最大迟到时间；

双流JOIN

public class JoinWindow {public static void main(String[] args) {StreamBuilder builder = new StreamBuilder("joinWindow");//左流RStream<User> user = builder.source("user", total -> {User user1 = JSON.parseObject(total, User.class);return new Pair<>(null, user1);});//右流RStream<Num> num = builder.source("num", source -> {Num user12 = JSON.parseObject(source, Num.class);return new Pair<>(null, user12);});//自定义join后的运算ValueJoinAction<User, Num, Union> action = new ValueJoinAction<User, Num, Union>() {@Overridepublic Union apply(User value1, Num value2) {...}};user.join(num).where(User::getName).equalTo(Num::getName).window(WindowBuilder.tumblingWindow(Time.seconds(30))).apply(action).print();TopologyBuilder topologyBuilder = builder.build();Properties properties = new Properties();properties.put(MixAll.NAMESRV_ADDR_PROPERTY, "127.0.0.1:9876");RocketMQStream rocketMQStream = new RocketMQStream(topologyBuilder, properties);rocketMQStream.start();}
}

双流聚合示例要点：

1、支持window join和非window join，对于非window join，只需要在上述及连表达式中去掉window即可；

2、支持多种窗口类型的window join；

3、支持对join后数据自定义操作；

03 参与贡献

RocketMQ Streams是Apache RocketMQ的子项目，已经在社区开源，参与RocketMQ Streams相关工作，请参考以下资源：

1、试用RocketMQ Streams，并阅读相关文档以了解更多信息；

maven仓库坐标：

<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-streams</artifactId><version>1.1.0</version>
</dependency>

RocketMQ Streams文档：

https://rocketmq.apache.org/zh/docs/streams/30RocketMQ%20Streams%20Overview

2、参与贡献：如果你有任何功能请求或错误报告，请随时提交 Pull Request 来分享你的反馈和想法；

社区仓库：

https://github.com/apache/rocketmq-streams

3、联系我们：可以在 GitHub上创建 Issue，向 RocketMQ 邮件列表发送电子邮件，或在 RocketMQ Streams SIG 交流群与专家共同探讨，RocketMQ Streams SIG加入方式：添加“小火箭”微信，回复RocketMQ Streams。

邮件列表：

https://lists.apache.org/list.html?dev@rocketmq.apache.org

RocketMQ Streams 1.1.0: 轻量级流处理再出发相关推荐

RocketMQ Streams拓扑构建与数据处理过程
本文作者:倪泽,Apache RocketMQ committer.RSQLDB/RocketMQ Streams Maintainer 01 背景 RocketMQ Streams 1.1.0版本已 ...
RocketMQ Streams在云安全及 IoT 场景下的大规模最佳实践
本文作者:袁小栋,Apache RocketMQ Committer,RocketMQ Streams Cofonder,阿里云安全智能计算引擎负责人 RocketMQ Streams简介 Rocke ...
mysql mgr_MySQL 8.0 MGR 流控功能
MGR(MySQL Group Replication)是MySQL集群技术的一个重大改进,而流控(Flow Control)在MGR组复制的性能和数据完整性上起着非常关键的作用,本文将解释MGR中的 ...
空指针：从 0 到 NULL，再到 nullptr
nullptr 空指针:从 0 到 NULL,再到 nullptr NULL 是一个宏定义: #undef NULL #if defined(__cplusplus) #define NULL 0 # ...
云计算情报局预告｜告别 Kafka Streams，让轻量级流处理更加简单
作者:不周关键词:Kafka ETL,高弹性.免运维.低成本阿里云消息队列 Kafka 版提供兼容 Apache Kafka 生态的全托管服务,彻底解决开源产品长期的痛点,是大数据生态中不可或缺的 ...
再见 2020！Apache RocketMQ 发布 4.8.0，DLedger 模式全面提升！
作者 | RocketMQ社区来源|阿里巴巴云原生公众号 "童年的雨天最是泥泞,却是记忆里最干净的曾经.凛冬散尽,星河长明,新的一年,万事顺遂,再见,2020!" 走过这个岁末, ...
利用EVC快速开发WINCE5.0的流驱动(转载)
WinCE5.0提供了一个标准的流驱动格式,大大方便了设备驱动程序的开发工作.但是传统的开发方式往往效率很低.方法如下: 1.在Platform Builder下建立一个流驱动的dll工程. 2.为流 ...
Apache RocketMQ 发布 v4.4.0，新添权限控制和消息轨迹特性
近日,Apache RocketMQ 发布了 v4.4.0,该版本主要增加了权限控制(ACL)和消息轨迹(Message Trace)两大特性,并做了8项优化,和修复了4处bug. 权限控制(ACL) ...
(连载)Android 8.0 : 系统启动流程之Linux内核
这是一个连载的博文系列,我将持续为大家提供尽可能透彻的Android源码分析 github连载地址前言 Android本质上就是一个基于Linux内核的操作系统,与Ubuntu Linux.Fedo ...

RocketMQ Streams 1.1.0: 轻量级流处理再出发

RocketMQ Streams 1.1.0: 轻量级流处理再出发相关推荐

最新文章

热门文章