您现在的位置是：首页 > 文章详情

Flink+ice 实现可视化规则编排与灵活配置(Demo)

日期：2022-09-16点击：281收藏

ice文档站：http://124.221.148.247/zh

1 Demo仓库地址：

github：https://github.com/zjn-zjn/flink-ice

gitee：https://gitee.com/waitmoon/flink-ice

2 Demo功能描述

通过netcat制造输入流（nc -l 9000 windows：nc -l -p 9000）

flink接收本地9000端口输入流，以回车(\n)分割单词

输入流经过IceProcessor处理后打印结果流

3 项目搭建

使用flink-quickstart-java快速搭建flink项目

3.1 添加ice依赖

因flink为非Spring项目，需依赖ice-core并手动初始化，Spring项目直接依赖ice-client-spring-boot-starter即可

<dependency> <groupId>com.waitmoon.ice</groupId> <artifactId>ice-core</artifactId> <version>${ice.version}</version> </dependency>

3.2 编写StreamingJob

public class StreamingJob { public static void main(String[] args) throws Exception { // 创建 Flink 执行环境 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //接收本地socket9000端口输入流，以回车分割单词 //通过netcat制造输入流 nc -l 9000 (windows nc -l -p 9000) DataStreamSource<String> stream = env.socketTextStream("localhost", 9000, "\n"); //按照单词长度keyBy，使用IceProcessor并打印结果 stream.keyBy(String::length).process(new IceProcessor()).print().setParallelism(1); //执行程序 env.execute("Flink Streaming Java API Skeleton"); } }

3.3 编写ice算子IceProcessor

在static代码块中初始化ice客户端，此处直接使用的自己部署的ice-server地址对应的app：2

算子功能： 将流内数据放入roam，组装pack并执行ice规则处理（直接根据iceId触发，iceId在server配置后台获取）

/** * ice算子 */ public class IceProcessor extends KeyedProcessFunction<Integer, String, String> { //ice 客户端 private static IceNioClient iceNioClient; static { //初始化ice客户端 try { //配置远程server地址，app，以及节点扫描路径 //此处使用了自己搭建的server，后台地址 http://eg.waitmoon.com/config/list/2 iceNioClient = new IceNioClient(2, "waitmoon.com:18121", "com.waitmoon.flink.ice.node"); //启动ice客户端 iceNioClient.start(); } catch (Exception e) { throw new RuntimeException(e); } } @Override public void processElement(String value, Context ctx, Collector<String> out) { //组装IcePack IcePack pack = new IcePack(); //设置要触发的iceId(配置后台中需要触发的ID) //http://eg.waitmoon.com/config/detail/2/1081 pack.setIceId(1081); //初始化roam，将单词和长度放入roam中 IceRoam roam = new IceRoam(); roam.put("input", value); roam.put("length", ctx.getCurrentKey()); pack.setRoam(roam); //同步执行 Ice.syncProcess(pack); //执行完成后，获取roam中的result String result = roam.getMulti("result"); if (result != null) { //result不为空，将结果放入下游算子 out.collect(result); } } @Override public void close() { if (iceNioClient != null) { //清理ice 客户端 iceNioClient.destroy(); iceNioClient = null; } } }

3.4 编写节点ContainsFlow

节点功能： 判断根据key去roam里拿的值是否在set中，是则返回true，否则返回false

/** * @author waitmoon * 过滤性质节点 * 判断值在不在集合中 */ @Data @Slf4j @EqualsAndHashCode(callSuper = true) public class ContainsFlow extends BaseLeafRoamFlow { //默认input private String key = "input"; private Set<String> set; @Override protected boolean doRoamFlow(IceRoam roam) { //判断roam中的key对应的值是否在集合中 return set.contains(roam.<String>getMulti(key)); } @Override public void afterPropertiesSet() { log.info("ContainsFlow init with key:{}, set:{} nodeId:{}", key, set, this.getIceNodeId()); } public NodeRunStateEnum errorHandle(IceContext ctx, Throwable t) { log.error("error occur id:{} e:", this.findIceNodeId(), t); return super.errorHandle(ctx, t); } }

3.5 编写节点PutNone

节点功能： 将value值放入roam的key中，不干扰流程(不返回true/false)

/** * @author waitmoon * 不干扰流程性质节点 * 将一个值放入roam */ @Data @EqualsAndHashCode(callSuper = true) public class PutNone extends BaseLeafRoamNone { //默认result private String key = "result"; private Object value; @Override protected void doRoamNone(IceRoam roam) { //将value放到roam中 roam.putMulti(key, value); } }