从零开始搭建flink流式计算项目-2小试牛刀-物联网场景下,如何实现设备采集参数监控报警功能

/\*\*

* 设备ID
*/
private Integer deviceId;

/\*\*

* 监控的变量名称
*/
private String varName;

/\*\*

* 最小值
*/
private Double min;

/\*\*

* 最大值
*/
private Double max;

}

##### 报警事件

/**
* 报警消息
*/
@Data
public class AlarmMessage {

/\*\*

* 设备
*/
private Integer deviceId;

/\*\*

* 报警时间
*/
private Long timestamp;
/**
* 触发报警的采集变量名称
*/
private String alarmVar;

/\*\*

* 触发报警的采集值
*/
private Number alarmValue;
}

#### 开始实现

public class IotMonitorJob {

public static void main(String[] args) throws Exception {


    StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
    environment.setParallelism(1);

    // 采集数据Stream
    DataStreamSource<IotData> iotDataStream = getIotStream(environment);
    // 报警规则Stream
    DataStreamSource<AlarmRule> ruleConfig = getRuleConfig(environment);
    // 缓存报警规则 并监控报警数据
    SingleOutputStreamOperator<AlarmMessage> alarmStream = iotDataStream.connect(ruleConfig)
            .keyBy(IotData::getDeviceId, AlarmRule::getDeviceId)
            .process(new CoProcessFunction<IotData, AlarmRule, AlarmMessage>() {

                // 用临时保存设备的报警规则 ,这里的状态交由flink维护
                private MapState<Integer, AlarmRule> alarmRuleValueState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    super.open(parameters);
                    // 初始化 ValueState
                    alarmRuleValueState = getRuntimeContext().getMapState(new MapStateDescriptor<>("alarm-rule-state", Integer.class, AlarmRule.class));
                }

                @Override
                public void processElement1(IotData iotData, CoProcessFunction<IotData, AlarmRule, AlarmMessage>.Context context, Collector<AlarmMessage> collector) throws Exception {
                    Map<String, Double> data = iotData.getData();

                    // 遍历每个规则
                    alarmRuleValueState.values().forEach(rule -> {

                        String varName = rule.getVarName();
                        // 获取变量值
                        Double val = data.get(varName);
                        if (val == null) {
                            // 变量里没有值
                            return;
                        }

                        if (val <= rule.getMin() || val > rule.getMax()) {
                            // 超过限制,输出报警信息
                            AlarmMessage alarmMessage = new AlarmMessage();
                            alarmMessage.setDeviceId(iotData.getDeviceId());
                            alarmMessage.setTimestamp(iotData.getTimestamp());
                            alarmMessage.setAlarmVar(varName);
                            alarmMessage.setAlarmValue(val);
                            collector.collect(alarmMessage);
                        }
                    });


                }

                @Override
                public void processElement2(AlarmRule alarmRule, CoProcessFunction<IotData, AlarmRule, AlarmMessage>.Context context, Collector<AlarmMessage> collector) throws Exception {
                    // 接收到AlarmRule, 仅更新 alarmRuleValueState
                    alarmRuleValueState.put(alarmRule.getId(), alarmRule);
                }
            });


    alarmStream.print();
    environment.execute();
}

/\*\*

* 获取物联采集数据
*
* @param environment
* @return
*/
private static DataStreamSource getIotStream(StreamExecutionEnvironment environment) {
return environment.addSource(new SourceFunction<>() {
private boolean running = true;

        @Override
        public void run(SourceContext<IotData> sourceContext) throws Exception {
            while (running) {

                // 模拟100个设备 每秒一次上报数据

                long ts = System.currentTimeMillis();
                ts = ts - ts % 1000;

                for (int i = 0; i < 100; i++) {
                    IotData iotData = new IotData();
                    iotData.setTimestamp(ts);
                    iotData.setDeviceId(i);

                    Map<String, Double> data = new HashMap<>();
                    data.put("var1", RandomUtils.nextDouble());
                    data.put("var2", RandomUtils.nextDouble());
                    iotData.setData(data);

                    sourceContext.collect(iotData);
                }

                Thread.sleep(1000 - ts % 1000);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    });
}


/\*\*

* 获取规则配置
*/
public static DataStreamSource getRuleConfig(StreamExecutionEnvironment environment) {
// 仅针对部分设备监控

    List<AlarmRule> ruleList = new ArrayList<>();
    for (int i = 0; i < 20; i++) {
        AlarmRule alarmRule1 = new AlarmRule();
        alarmRule1.setDeviceId(i);
        alarmRule1.setVarName("var1");
        alarmRule1.setMax(20.0);
        alarmRule1.setMin(0.0);
        ruleList.add(alarmRule1);

        AlarmRule alarmRule2 = new AlarmRule();
        alarmRule2.setDeviceId(i);
        alarmRule2.setVarName("var2");
        alarmRule2.setMax(10.0);
        alarmRule2.setMin(0.0);
        ruleList.add(alarmRule2);

    }
    return environment.fromCollection(ruleList);
}

}

### 启动job



> 
> 实际运行基于 java 11 , flink 1.18.1
> 
> 
> 


输出结果:

AlarmMessage(deviceId=0, timestamp=1709732511000, alarmVar=var2, alarmValue=1.0408785873261203E308)
AlarmMessage(deviceId=1, timestamp=1709732511000, alarmVar=var2, alarmValue=8.409717342118433E306)
AlarmMessage(deviceId=2, timestamp=1709732511000, alarmVar=var2, alarmValue=6.955367711979709E307)
AlarmMessage(deviceId=3, timestamp=1709732511000, alarmVar=var2, alarmValue=2.5403069646236554E307)
AlarmMessage(deviceId=4, timestamp=1709732511000, alarmVar=var2, alarmValue=7.629789041713245E307)
AlarmMessage(deviceId=5, timestamp=1709732511000, alarmVar=var2, alarmValue=6.918664964996954E307)
AlarmMessage(deviceId=6, timestamp=1709732511000, alarmVar=var2, alarmValue=1.1660434456728436E308)
AlarmMessage(deviceId=7, timestamp=1709732511000, alarmVar=var2, alarmValue=2.1272561368179368E307)
AlarmMessage(deviceId=8, timestamp=1709732511000, alarmVar=var2, alarmValue=2.8693117885744695E307)
AlarmMessage(deviceId=9, timestamp=1709732511000, alarmVar=var2, alarmValue=1.1232501067396574E308)
AlarmMessage(deviceId=10, timestamp=1709732511000, alarmVar=var2, alarmValue=1.6192738031099514E308)
AlarmMessage(deviceId=11, timestamp=1709732511000, alarmVar=var2, alarmValue=7.515829766654446E307)
AlarmMessage(deviceId=12, timestamp=1709732511000, alarmVar=var2, alarmValue=1.6409410780574847E308)
AlarmMessage(deviceId=13, timestamp=1709732511000, alarmVar=var2, alarmValue=7.372363635115241E307)
AlarmMessage(deviceId=14, timestamp=1709732511000, alarmVar=var2, alarmValue=5.269385013806783E306)
AlarmMessage(deviceId=15, timestamp=1709732511000, alarmVar=var2, alarmValue=9.736804956554577E307)
AlarmMessage(deviceId=16, timestamp=1709732511000, alarmVar=var2, alarmValue=5.403962718372102E307)
AlarmMessage(deviceId=17, timestamp=1709732511000, alarmVar=var2, alarmValue=1.7957965318588386E308)
AlarmMessage(deviceId=18, timestamp=1709732511000, alarmVar=var2, alarmValue=6.546384330721207E307)
AlarmMessage(deviceId=19, timestamp=1709732511000, alarmVar=var2, alarmValue=1.2797848722222382E308)
AlarmMessage(deviceId=0, timestamp=1709732512000, alarmVar=var2, alarmValue=8.096850966966417E307)
AlarmMessage(deviceId=1, timestamp=1709732512000, alarmVar=var2, alarmValue=1.1459880504481993E308)
AlarmMessage(deviceId=2, timestamp=1709732512000, alarmVar=var2, alarmValue=1.6878563127635106E308)
AlarmMessage(deviceId=3, timestamp=1709732512000, alarmVar=var2, alarmValue=1.3431398337246118E308)
自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数嵌入式工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年嵌入式&物联网开发全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上嵌入式&物联网开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新

如果你觉得这些内容对你有帮助，可以+V：Vip1104z获取！！！ （备注：嵌入式）

最后

资料整理不易，觉得有帮助的朋友可以帮忙点赞分享支持一下小编~

你的支持，我的动力；祝各位前程似锦，offer不断，步步高升！！！

项目、讲解视频，并且后续会持续更新**

如果你觉得这些内容对你有帮助，可以+V：Vip1104z获取！！！ （备注：嵌入式）

最后

资料整理不易，觉得有帮助的朋友可以帮忙点赞分享支持一下小编~

你的支持，我的动力；祝各位前程似锦，offer不断，步步高升！！！

更多资料点击此处获qu！！文章来源地址https://www.toymoban.com/news/detail-848193.html

到了这里，关于从零开始搭建flink流式计算项目-2小试牛刀-物联网场景下,如何实现设备采集参数监控报警功能的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！