物联网的发展为智能安防和自动化监控带来了更多便利，同时，新型城镇建设、智慧城市与智慧社区的发展也为门禁管理等安防问题智能化提出了更高的要求。在智能化发展的背景下，门禁成为一套集成了访客、考勤、消费、巡更、梯控等更多功能的全面便捷的系统安全应用，随着门禁系统应用愈发广泛，对海量数据的实时快速处理也成为了日益重要的问题。
作为一款高性能分布式时序数据库，DolphinDB 提供了流数据表和流计算引擎用于实时数据处理，为智能安防提供了有力支持。今天，我们将介绍如何通过流计算引擎多级级联实现对门禁设备异常状态的实时监测，基于 DolphinDB 流数据处理框架，为大家提供一套实时监测门禁设备异常状态的低延时解决方案。
常见的智能安防系统一般包含有监控、报警、门禁和远程控制4个主要功能，可以单独运行也可统一管理。而门禁则是整个智能安防系统中的基础应用，同时也关系到物联网领域中的公共安全、城市管理、智能家居等多个方面。首先，我们先来了解一下门禁系统的具体应用场景与功能设置：
1.门禁、报警综合管理系统服务器：提供集中管理及监控、输出、联动功能。
2.门禁工作站：门禁工作站提供功能设置及事件监控。
3.门禁控制器：门禁管理系统的核心部分，直接管理及控制相关设备，具有存储功能，可存放持卡人资料及各种事件记录。
4.读卡器：工作于射频方式，采集感应卡的数据传输到门禁控制器，以便控制器进行各种管理及相应的控制。
5.电锁：电子方式开关，实现开门及锁门，由门禁控制器直接控制。
6.开门按钮：提供便利的开门方式。
7.门磁：检测门的状态信息，然后传输到控制器。
8.报警输入输出设备：为加强系统的保安，可以将输入输出设备接入门禁控器的接口，实现系统的报警及联动。
工业中心门禁管理系统结构
上图展示了一个常见的门禁管理系统结构，报警系统是安防及门禁系统中保障安全问题的重要功能组件及环节，也可以实现与其他监控设备的联防联控。
1. 设计思路与方案
随着接入智能门禁系统的终端不断增多，如何对海量数据进行实时高效计算、及时反馈报警消息，成为智能门禁及智慧社区建设的关键问题。保持门禁正常关闭状态是保证楼宇内居民安全的基础需求之一，假定有一个监控系统，对所有门禁设备每5秒钟采集1次数据，同时开门或关门的事件会主动上报数据，本案例中的门禁异常状态检测需求是：开门状态连续存在超过5分钟报警。首先需要对采集数据进行去重处理，再检测出数据中状态持续超时的记录。此时的记录会包括所有状态持续超过5分钟的数据，因此仍需将数据接入下一级引擎去除关门告警，只保留开门状态超时报警。采用 DolphinDB 响应式状态引擎完成过滤筛选数据的任务，并通过会话窗口引擎检测超时数据。将三个引擎级联，实现检测异常门禁状态(开门时间大于5分钟)的流水线处理模式。
门禁异常状态数据处理流程
2. 内置流计算引擎监测门禁异常状态
DolphinDB 内置了响应式状态引擎和会话窗口引擎实现门禁异常状态的实时监测。
响应式状态引擎(createReactiveStateEngine)：
·有两种因子计算：无状态因子和有状态因子。
·无状态因子只依赖最新数据，不需要历史数据。
·有状态因子需要最新数据和历史数据，并会更新状态。
·每输入一条数据都将触发一条结果输出，输入和输出数据量一致。
·算子中只能包含向量函数，对于常见状态算子进行了优化，大幅提高计算效率。
会话窗口引擎(creatSessionWindowEngine)：
·适用于物联网场景，解决设备在线时间不均匀的问题。
·窗口不按固定频率生成，窗口长度也不是固定的。
·以第一条数据的时间戳作为第一个窗口的起始时间。
·若在指定等待时间内未收到新数据，则当前窗口结束。
·解决滑动窗口计算中无数据窗口增加不必要计算开销的问题。
3. 实现步骤
接下来，我们一起来详细了解下如何通过 DolphinDB 实现多级引擎级联检测异常门禁状态的流水线处理模式。
3.1 定义并共享输入输出流数据表
首先定义一个用于实时接收门禁监控设备数据的流数据表，表结构共包含七列，通过 enableTableShareAndPersistence函数共享流数据表并持久化到硬盘上。其次定义异常状态流数据表，用于响应式状态引擎的输出，并将其持久化到磁盘上。
3.2 创建响应式状态引擎过滤重复数据
响应式状态引擎会对输入的每一条消息做出计算响应，产生一条记录作为结果，可以启用过滤条件保证只有满足条件的结果才会输出。在本案例中，只有事件码有变化的数据才会被输出到设置的输出表中。DolphinDB 内置的流计算引擎均实现了数据表的接口，因此多个引擎流水线处理变得异常简单，只要将后一个引擎作为前一个引擎的输出即可。
3.3 通过级联会话窗口引擎检测状态超时数据
首先创建一张内存表，为响应式状态引擎提供输入的表结构，与上一级引擎输出表的结构一致。在会话窗口引擎中，设置分组列 keyColumn 为门号 doorNum ，时间列 timeColumn 为时间 eventDate 。检测需求是五分钟内无数据报警，因此 sessionGap 为300(秒)，表示收到某条数据后经过该时间的等待仍无新数据到来，就终止当前窗口。 metrics 设为 last(doorEventCode) ，即返回窗口内的最后一条记录数据。会话窗口引擎的输入数据为上一级响应式状态引擎的输出，输出作为下一级响应式状态引擎的输入。
3.4 响应式状态引擎过滤关门告警
上级会话窗口引擎获取到的数据包括开门和关门超过5分钟的数据，因此需要再通过响应式状态引擎过滤掉关门状态超时数据，只保留开门告警。
3.5 订阅流数据
过滤了关门告警数据后，订阅流数据表 doorRecord 并把收到的流数据写入上述会话窗口引擎。
3.6 从 MQTT 服务器接收数据
DolphinDB 提供了 MQTT 插件用于订阅 MQTT 服务器的数据，在订阅时需要数据格式解析函数，目前插件提供了 json 和 csv 格式的解析函数，本例使用 mqtt::createJsonParser 解析 json 格式数据。随着网络与数字技术飞速发展，门禁系统早已不再是单纯的门道及钥匙管理，而是逐渐演变成为一套完整的出入门禁安全管理系统，集微机自动识别技术和现代安全管理措施为一体。DolphinDB 提供了一种实时监测门禁设备异常状态的低延时解决方案，实现了会话窗口引擎和响应式状态引擎级联，降低开发难度，有效提升海量数据的实时计算效率，满足了门禁系统智能化的计算需求。
以上是基于 DolphinDB 流数据处理框架进行引擎级联监测门禁异常状态的完整方案，想知道详细实现步骤并自己动手模拟验证?欢迎关注公众号【DolphinDB 物联网】深入了解。