数据处理方法、驱动中心节点及物联网驱动中心系统与流程

本发明涉及物联网，具体涉及一种数据处理方法、驱动中心节点及物联网驱动中心系统。

背景技术：

1、相关技术中，在物联网（internet of things，iot）云对云数据采集领域，物联网数据处理平台通常采用静态编译驱动架构，处理数据的接口与核心框架强耦合，各个平台接口的驱动jar需随主程序静态编译部署，新增或修改驱动需重启服务，且消息数据处理采用全局固定单体线程池处理所有消息数据，存在易造成线程阻塞、资源浪费和系统扩展性不佳等问题。

技术实现思路

1、本发明提供了一种数据处理方法、驱动中心节点及物联网驱动中心系统，旨在能够解决相关技术的物联网数据处理平台的线程阻塞、资源浪费和系统扩展性不佳等技术问题。

2、第一方面，本发明提供了一种基于物联网驱动中心系统的数据处理方法，应用于驱动中心节点，所述物联网驱动中心系统包括所述驱动中心节点和消息队列集群，所述数据处理方法包括：

3、从消息队列集群中消费目标消息主题当前的待处理消息，所述待处理消息包括待处理数据；

4、根据所述目标消息主题与所述待处理消息的功能类型，从当前的处理器路由映射关系中匹配对应的目标处理单元，所述处理器路由映射关系用于记录各消息主题与消息功能类型的组合与当前驱动jar中各处理单元之间的映射关系；

5、根据所述目标消息主题，将所述待处理消息的数据处理任务分发至与所述目标消息主题绑定的目标线程池执行；

6、其中，所述目标线程池用于根据所述数据处理任务执行所述目标处理单元的数据处理逻辑。

7、在一些实施例中，当前驱动jar通过以下方式进行驱动加载更新：

8、响应于接收到驱动配置变更事件的消息，根据消息中的驱动标识与版本信息从数据库中读取当前驱动元数据，所述当前驱动元数据包括当前驱动版本、远程下载地址和校验值；

9、从远程下载地址下载当前驱动jar，并根据校验值进行完整性校验；

10、响应于当前驱动jar校验通过，创建对应的当前类加载器；

11、通过当前类加载器加载当前驱动jar，实例化当前驱动jar包中的各处理单元；

12、根据当前驱动jar包中的各处理单元更新当前的处理器路由映射关系。

13、在一些实施例中，在所述根据当前驱动jar包中的各处理单元更新当前的处理器路由映射关系之后，所述方法还包括：

14、对历史驱动jar对应的历史类加载器进行过期标记，并在历史驱动jar对应的待执行任务执行完毕后延迟回收历史类加载器。

15、在一些实施例中，在所述根据所述目标消息主题，将所述待处理消息的数据处理任务分发至与所述目标消息主题绑定的目标线程池执行之后，所述方法还包括：

16、通过守护线程定时检查各数据处理任务的执行状态；

17、对超时任务进行中断处理并释放对应的处理线程资源；

18、对执行异常任务进行重试执行，或者，将执行异常任务对应的待处理消息转入消息队列集群中预设的异常消息主题中。

19、在一些实施例中，在所述通过守护线程定时检查各数据处理任务的执行状态之前，所述方法还包括：将各数据处理任务的任务执行上下文写入超时任务映射表，并记录对应的消息主题以及对应的待处理消息在消息队列集群中的缓存位置；

20、所述通过守护线程定时检查各数据处理任务的执行状态，包括：

21、通过守护线程遍历超时任务映射表中各数据处理任务的任务执行上下文，确定各数据处理任务的执行状态，并判断各数据处理任务是否超时执行；

22、将各数据处理任务的执行状态记录至数据库中。

23、在一些实施例中，在所述从当前的处理器路由映射关系中匹配对应的目标处理单元之后，所述方法还包括：

24、响应于未在处理器路由映射关系中匹配到对应的目标处理单元，将所述待处理消息标记为异常消息并记录至数据库。

25、在一些实施例中，所述物联网驱动中心系统还包括驱动网关，在所述从消息队列集群中消费目标消息主题当前的待处理消息之前，所述方法还包括：

26、所述驱动网关获取待处理数据，并将待处理数据转化为消息投递至消息队列集群。

27、第二方面，本发明还提供一种驱动中心节点，所述驱动中心节点包括：

28、消息消费模块，用于从消息队列集群中消费目标消息主题当前的待处理消息，所述待处理消息包括待处理数据；

29、处理器路由模块，用于根据所述目标消息主题与所述待处理消息的功能类型，从当前的处理器路由映射关系中匹配对应的目标处理单元，所述处理器路由映射关系用于记录各消息主题与消息功能类型的组合与当前驱动jar中各处理单元之间的映射关系；

30、消息处理模块，用于根据所述目标消息主题，将所述待处理消息的数据处理任务分发至与所述目标消息主题绑定的目标线程池执行；

31、其中，所述目标线程池用于根据所述数据处理任务执行所述目标处理单元的数据处理逻辑

32、在一些实施例中，所述驱动中心节点还包括动态类加载模块；

33、所述动态类加载模块用于执行当前驱动jar的驱动加载更新过程，所述驱动加载更新过程包括：

34、响应于接收到驱动配置变更事件的消息，根据消息中的驱动标识与版本信息从数据库中读取当前驱动元数据，所述当前驱动元数据包括当前驱动版本、远程下载地址和校验值；

35、从远程下载地址下载当前驱动jar，并根据校验值进行完整性校验；

36、响应于当前驱动jar校验通过，创建对应的当前类加载器；

37、通过当前类加载器加载当前驱动jar，实例化当前驱动jar包中的各处理单元；

38、根据当前驱动jar包中的各处理单元更新当前的处理器路由映射关系。

39、第三方面，本发明还提供一种物联网驱动中心系统，所述物联网驱动中心系统包括：

40、驱动网关，用于维护管理系统驱动配置以及采集待处理数据，并将驱动配置变更事件和待处理数据转化为消息投递至消息队列集群；

41、驱动中心节点，用于执行第一方面的任一实施例所提供的数据处理方法；

42、消息队列集群，用于管理各消息主题下的消息，包括待处理数据对应的待处理消息和驱动配置变更事件的消息；

43、数据库，用于存储当前最新的驱动配置、驱动网关采集的待处理数据和驱动中心节点的数据处理结果。

44、根据本发明实施例提供的基于物联网驱动中心系统的数据处理方法，在消费消息队列集群中的待处理消息后，按照待处理消息的消息主题和功能类型，通过预先配置的处理器路由映射关系将待处理消息路由映射到对应的目标处理单元，不同主题、不同功能类型的待处理消息可以直接路由到各自匹配的目标处理单元进行处理，即为不同业务、不同设备的异构数据消息精准定位匹配对应的处理单元进行数据处理，统一了异构数据消息的路由入口，适配多协议、多品类iot设备，屏蔽不同数据消息的底层协议差异，使得系统具备良好的通用性和兼容扩展能力；同时，预先通过将不同消息主题分别与对应的独立线程池对应绑定，待处理消息匹配到目标处理单元后，将待处理消息的数据处理任务分发至与目标消息主题绑定的目标线程池执行，目标线程池根据数据处理任务执行目标处理单元的数据处理逻辑，实现了线程资源按主题隔离，不同主题互不干扰，有利于针对各主题按需调整对应的线程资源，有利于支撑主题流量波动，降低消息堆积风险，有效改善单主题流量突增导致全局线程阻塞和线程争抢的问题，有利于改善系统资源浪费问题，提升系统资源利用率和系统稳定性。