CN115237987A - 车辆多源异构数据的接入和处理方法、模块及系统 - Google Patents

Abstract

本发明公开了一种车辆多源异构数据的接入和处理方法、模块及系统。本发明的接入方法包括：当有车端、路侧端或第三方平台发送数据接入请求时，判断数据接入请求所采用的协议；调用所采用的协议对应的协议处理器读取当前协议数据；将接收的协议数据进行反序列化；将反序列化的消息投递到disruptor消息队列；批量将disuptor队列中的数据放入kafka集群的kafka消息队列，实现数据接入。本发明可以以实现多源异构数据的接入和处理，满足各类汽车、交通基础数据的融汇管理。

Description

车辆多源异构数据的接入和处理方法、模块及系统

技术领域

本发明涉及数据接入和处理技术，尤其涉及一种车车辆多源异构数据的接入和处理方法、模块及系统。

背景技术

随着智能网联汽车产业的发展，云控平台的支撑优势和效果将愈发显著，其发展主要为满足行业、技术、示范应用以及标准法规建设等方面的产业需求。

在标准层面上，现有数据标准难以满足平台采集和处理加工的需求。目前，车辆动态数据、车路协同数据、道路基础设施数据和交通管理数据有限，难以满足汽车大数据云控基础平台对车辆自动驾驶辅助的需求。而且目前缺乏对终端数据、各级平台数据的接入与融合做引导或强制性标准要求，导致数据异构，因而云控平台的基础数据获取难度较大。

现有整车企业平台是企业根据各自需求而建设的，缺乏统一标准和信息交互设计，导致很多不同品牌网联车、不同路侧品牌、以及各运营第三方平台的各协议、数据规格均不一样，成为智能网联汽车发展的障碍，不利于标准互联互通和行业管理。为了实现车与车之间、车与路之间、路与路之间、云与云之间、车与云之间、路与云之间不同品牌车互通互认，需解决中心云的多源异构数据接入、处理问题。

发明内容

发明目的：本发明针对现有技术存在的问题，提供一种车辆多源异构数据的接入和处理方法、模块及系统。

技术方案：本发明所述的车辆多源异构数据接入方法包括：

当有车端、路侧端或第三方平台发送数据接入请求时，判断数据接入请求所采用的协议；

调用所采用的协议对应的协议处理器读取当前协议数据；

将接收的协议数据进行反序列化；

将反序列化的消息投递到disruptor消息队列；

批量将disuptor队列中的数据放入kafka集群的kafka消息队列，实现数据接入。

进一步的，通过开放的socket接口实现数据接入请求的监听。

进一步的，所述将接收的协议数据进行反序列化，具体包括：

读取接收的协议数据的数据体，通过解析得到数据体的数据格式、数据主题；

将解析得到的数据格式、数据主题和数据体合并，重新封装成一个消息，实现反序列化。

进一步的，该方法还包括：若由kafka集群向disruptor消息队列返回了消息，则将disruptor消息队列的消息序列化后，通过对应的协议处理器向数据接入请求方返回数据

进一步的，所述将disruptor消息队列的消息序列化，具体包括：

解析disruptor消息队列的消息，得到消息中包含的数据格式、数据主题和数据体；

将数据体按照对应协议形成协议数据。

本发明所述的车辆多源异构数据处理方法，是对于按照上述方法接入的数据进行数据处理，该方法具体包括：

接收kafka集群的kafka消息队列发送的消息；

解析接收到的消息，得到消息中包含的数据体以及数据体的数据格式、数据主题；

根据解析得到的数据格式，调用对应的数据格式解析器对数据体进行解析，得到统一格式的数据内容；

将数据内容按照对应数据主题的字段转换配置对应关系，重新封装成统一预设数据结构的数据；

将统一预设数据结构的数据投递到disruptor消息队列。

本发明所述的车辆多源异构数据接入模块包括：

协议判断单元，用于当有车端、路侧端或第三方平台发送数据接入请求时，判断数据接入请求所采用的协议；

协议数据读取单元，用于调用所采用的协议对应的协议处理器读取当前协议数据；

反序列化单元，用于将接收的协议数据进行反序列化；

disruptor消息队列，用于暂存反序列化的消息；

批量队列消息投递单元，用于批量将disuptor队列中的数据放入kafka集群的kafka消息队列；

以及kafka集群。

进一步的，所述接入模块还包括：

序列化单元，用于在kafka集群向disruptor消息队列返回了消息时，将disruptor消息队列的消息序列化，使得序列化后的通过对应的协议处理器向数据接入请求方返回数据。

本发明所述的车辆多源异构数据处理模块包括：

数据解析单元，用于解析接收到的消息，得到消息中包含的数据体以及数据体的数据格式、数据主题；

数据格式解析单元，用于根据解析得到的数据格式，调用对应的数据格式解析器对数据体进行解析，得到统一格式的数据内容；

数据封装单元，用于将数据内容按照对应数据主题的字段转换配置对应关系，重新封装成统一预设数据结构的数据；

disruptor消息队列，用于暂存统一预设数据结构的数据。

本发明所述的车辆多源异构数据的中心云处理系统包括：

上述车辆多源异构数据接入模块；

上述车辆多源异构数据处理模块；以及

数据存储模块，包括存储实时数据的memcache和redis数据库，和存储历史数据的ClickHouse数据库。

有益效果：本发明与现有技术相比，其显著优点是：本发明可以实现多源异构数据的接入和处理，满足各类汽车、交通基础数据的融汇管理，统一了数据服务标准，实现大范围的数据接入，提升数据利用率，降低互操作成本，连通信息孤岛，解闱了不同车路云之间的标准化通信协议与高效数据共享接口。

附图说明

图1是本发明实施例一提供的车辆多源异构数据接入方法的流程示意图；

图2是本发明实施例一提供的车辆多源异构数据接入方法的框架示意图；

图3是本发明实施例二提供的车辆多源异构数据处理方法的流程示意图；

图4是本发明实施例二提供的车辆多源异构数据处理方法的框架示意图；

图5是本发明实施例五提供的车辆多源异构数据中心云处理系统的框架示意图。

具体实施方式

实施例一

多源异构动态数据，包括车端、路侧端、平台端数据，其中车端数据包含车辆运行数据，车辆动力数据，车辆燃油数据，车辆预警等数据，路侧数据包括路侧设备的融合感知数据、激光雷达数据、道路交通参与者数据等，第三方平台数据包括车厂、车辆示范区、气象局、交管局等数据，在这样多数据接入情况下，很容易造成数据接收不及时，消息积压，甚至消息丢失，对于使用消费数据端造成一系列的错误，因此所以需要解决消息实时、可靠的接入。针对此问题，本实施例提供了一种车辆多源异构数据接入方法，如图1和图2所示，该方法可以包括：

S1、当有车端、路侧端或第三方平台发送数据接入请求时，判断数据接入请求所采用的协议。

多源异构数据的传输协议包含TCP/IP、UDP、MQ、Kafka、Http等多种协议，车企、示范区的传输的数据格式、对接的数据协议都不一样，因此会提供不同的协议对接接口。监听开放的socket接口接收数据，当监听到有车端、路侧端或第三方平台发送数据接入请求时，通过对接收的数据进行特定标记判断，判断发送的数据协议，目前主要的协议有TCP/IP、MQ、Http协议。

S2、调用所采用的协议对应的协议处理器读取当前协议数据。

若采用的协议为TCP/IP协议，则采用TCP/IP协议处理器读取二进制数据，若采用的协议为MQ协议，则采用采用MQ协议处理器读取MQ消息，若采用的协议为Http协议，则采用Http协议处理器读取HTTP消息。

S3、将接收的协议数据进行反序列化。

反序列化的具体步骤为：A读取接收的协议数据的数据体，通过解析得到数据体的数据格式、数据主题；其中，协议数据的数据格式包括xml、json、yml等，数据主题通过特定的数据位置可以解析得到；B、将解析得到的数据格式、数据主题和数据体合并，重新封装成一个消息，实现反序列化，具体是封装成一个json对象，在json对象里面包含数据格式、数据主题和数据体，json对象的格式具体如下：

其中，format指定“data”数据格式，比如xml、yml、json、binary等。topic指定数据主题，主要起标记作用，Data是传输的具体数据，即数据体。

S4、将反序列化的消息投递到disruptor消息队列。

disruptor队列是一个高速缓存的环形消息队列，在单机内缓存和消费数据。

S5、批量将disuptor队列中的数据放入kafka集群的kafka消息队列，实现数据接入。

通过批量消费Disuptor队列数据数据，批量将消息放入kafka消息队列，而不是一次放一条数据，这样减少了网络交互次数，提高了入队效率。

S6、若由kafka集群向disruptor消息队列返回了消息，则将disruptor消息队列的消息序列化后，通过对应的协议处理器向数据接入请求方返回数据

其中，序列化具体包括：解析disruptor消息队列的消息，得到消息中包含的数据格式、数据主题和数据体；将数据体按照对应协议形成协议数据。

本实施例对外提供TCP/IP、MQ、HTTP协议的接入，用来满足不同协议，不同来源的数据接入，后续可以依据实际增加对接协议。为了满足高效、实时，防止接入数据超时，所以在接入系统里面不做任何的数据计算。接入的数据不直接放入消息队列，而是暂时存放在Disuptor队列里面，因为接入数据直接放入消息队列里面，通过网络传输而消耗时间更长，直接放入Disuptor队列是一个内存队列。通过批量消费Disuptor队列数据数据，批量将消息放入消息队列，而不是一次放一条数据，这样减少了网络交互次数，提高了入队效率。

实施例二

如果处理高精数据的实时处理不及时，就会造成消息积压，虽然有消息队列存储实时数据，不会丢失数据，但是经过日积月累后，未处理消息会由少变多，失去了消息的实时性，造成消息的延时，针对此，本实施例提供了一种车辆多源异构数据处理方法，该方法是对于按照实施例一方法接入的数据进行数据处理，如图3和图4所示，该方法具体包括：

S1、接收kafka集群的kafka消息队列发送的消息。

S2、解析接收到的消息，得到消息中包含的数据体以及数据体的数据格式、数据主题。

传输的消息具体为json对象，其内容包括了数据体以及数据体的数据格式、数据主题，通过解析即可得到。数据格式可能为json格式、二进制格式或XML格式等。

S3、根据解析得到的数据格式，调用对应的数据格式解析器对数据体进行解析，得到统一格式的数据内容。

若解析得到的数据体的数据格式为json格式，则采用json解析器，对数据体进行解析，得到统一格式的数据内容，若解析得到的数据体的数据格式为二进制格式，则采用二进制解析器，对数据体进行解析，得到统一格式的数据内容，若解析得到的数据体的数据格式为XML格式，则采用XML解析器，对数据体进行解析，得到统一格式的数据内容。这里统一格式可以是任意一种通用的业务数据模型，不具体进行限定。

S4、将数据内容按照对应数据主题的字段转换配置对应关系，重新封装成统一预设数据结构的数据。

其中，预设数据结构可以是任意一种通用数据结构。

S5、将统一预设数据结构的数据投递到disruptor消息队列。

为了提高数据消息效率，批量把数据保存到存储模块。

本实施例实现实时动态数据处理计算，是把数据进行转换对象和把转换后的对象进行存储或者推入消息队列，进行下一步的消费处理。消息转换主要是把消息队列里面取出来的字节、json字符、xml字符转换成系统所需要的数据对象，对象转换是基于转换引擎实现，其本质是通过系统配置来实现转换的调度工作。为了高效的消费数据，数据处理模块引入Disrupor框架，存储转换后的对象。转换后的对象从Disrupor消费后，批量把数据保存到存储系统，提高数据消息效率。当计算模块出现瓶颈后，可以动态增加计算模块，实现多节点消费kafka的数据，配合kafka的分片机制和消费者组特点，进行数据的有序计算。

实施例三

本实施例提供了一种车辆多源异构数据接入模块，包括：

协议判断单元，用于当有车端、路侧端或第三方平台发送数据接入请求时，判断数据接入请求所采用的协议；

协议数据读取单元，用于调用所采用的协议对应的协议处理器读取当前协议数据；

反序列化单元，用于将接收的协议数据进行反序列化；

disruptor消息队列，用于暂存反序列化的消息；

批量队列消息投递单元，用于批量将disuptor队列中的数据放入kafka集群的kafka消息队列；

kafka集群；

序列化单元，用于在kafka集群向disruptor消息队列返回了消息时，将disruptor消息队列的消息序列化，使得序列化后的通过对应的协议处理器向数据接入请求方返回数据。

其中，本模块内除kafka集群外的所有单元都可通过动态数据接收网关实现。

该模块可采用软件和/或硬件的方式实现，该装置可以配置于终端设备中。

本实施例所提供的模块可以用于执行实施例一所提供的方法，具备执行方法相应的功能和有益效果。与实施例一接入方法一一对应，未详尽之处参照实施例一。

值得注意的是，上述模块置的实施例中，所包括的各个单元只是按照功能逻辑进行划分的，但并不局限于上述的划分，只要能够实现相应的功能即可；另外，各功能单元的具体名称也只是为了便于相互区分，并不用于限制本发明的保护范围。

实施例四

本实施例提供了一种车辆多源异构数据处理模块，包括：

数据解析单元，用于解析接收到的消息，得到消息中包含的数据体以及数据体的数据格式、数据主题；

数据格式解析单元，用于根据解析得到的数据格式，调用对应的数据格式解析器对数据体进行解析，得到统一格式的数据内容；

数据封装单元，用于将数据内容按照对应数据主题的字段转换配置对应关系，重新封装成统一预设数据结构的数据；

disruptor消息队列，用于暂存统一预设数据结构的数据。

该模块可采用软件和/或硬件的方式实现，该装置可以配置于终端设备中。

本实施例所提供的模块可以用于执行实施例二所提供的方法，具备执行方法相应的功能和有益效果。与实施例二方法一一对应，未详尽之处参照实施例一。

值得注意的是，上述模块置的实施例中，所包括的各个单元只是按照功能逻辑进行划分的，但并不局限于上述的划分，只要能够实现相应的功能即可；另外，各功能单元的具体名称也只是为了便于相互区分，并不用于限制本发明的保护范围。

实施例五

本实施例提供了一种车辆多源异构数据的中心云处理系统，如图5所示，包括：

实施例三所述的车辆多源异构数据接入模块；

实施例四所述的车辆多源异构数据处理模块；以及

数据存储模块，包括存储实时数据的memcache和redis数据库，和存储历史数据的ClickHouse数据库。

其中农，实时数据存储要提供数据的高性能使用，就采用内存数据库进行存储，内存数据库有memcache和redis数据库，memecache把数据全部存在内存之中，断电后会挂掉，数据不能超过内存大小，redis支持集群机制，可以横向扩展，同时redis的持久化机制有部份存在硬盘上，这样能保证数据的持久性，保障数据不丢失，redis在数据类型支持上要比memecache多很多，所以使用redis作为实时数据存储中间件。Redis里面存储的数据存储的是车辆的实时信息，采用redis的Hash结构存储车辆实时信息。存储格式如下表所示：

表1

历史数据存储采用ClickHouse实现。因为车端的数据非常大、种类也非常多，可达千亿级别数据，但是种类之间的关系比较简单，如果用传统的关系型数据库，基本无法满足要求，所以只能从分布式数据库考虑，目前比较流行的分布式数据库ClickHouse就非常适合。历史数据存储优点如下：写入速度快，50-200M/S,对于大量的数据更新非常实用。数据压缩空间大，减少IO，处理查询高吞吐量，每台服务器秒级数十亿行。查询快，比Vertica快5倍以上，比GP快10倍以上，比HIVE快200倍以上，比MYSQL快800倍以上。高效实用CPU,并行处理单个查询，充分利用多核，在多个服务器山分布式处理。开源的列存储，支持线性扩展，简单方便，高可用容错。大部分的SQL语句语法与采用标准的SQL语句一样，上手使用特别容易。

本实施例的中心云系统接入车端、路侧端、第三方平台的多源异构数据,包含TCP、UDP、MQTT、Kafka、Http等多种协议，通过动态数据接收网关进行协议适配认证接收数据之后，把消息推送到kafka模块进行缓存，通过消息队列的消息消费者把数据拉取出来进行综合计算之后，把计算之后的数据后分别存储在Redis实时缓存库，用于数据实时查询和Kafka消息机制，在数据处理模块通过规则引擎按预定义好的车辆登入登出、车辆运行、车辆告警、燃料数据、路-车辆运行状态、路-交通参与者、路-交通事件、信号灯数据、视频数据等数据规格进行数据协议解析、转换、统一的格式转换，数据转换成通用格式后，把数据推送到KafKa消息总线系统，Kafka消息总线推送给消费者数据中台做历史存储使用。

以上所揭露的仅为本发明较佳实施例而已，不能以此来限定本发明之权利范围，因此依本发明权利要求所作的等同变化，仍属本发明所涵盖的范围。

Claims (10)

1.一种车辆多源异构数据接入方法，其特征在于该方法包括：

当有车端、路侧端或第三方平台发送数据接入请求时，判断数据接入请求所采用的协议；

调用所采用的协议对应的协议处理器读取当前协议数据；

将接收的协议数据进行反序列化；

将反序列化的消息投递到disruptor消息队列；

批量将disuptor队列中的数据放入kafka集群的kafka消息队列，实现数据接入。

2.根据权利要求1所述的车辆多源异构数据接入方法，其特征在于：通过开放的socket接口实现数据接入请求的监听。

3.根据权利要求1所述的车辆多源异构数据接入方法，其特征在于：所述将接收的协议数据进行反序列化，具体包括：

读取接收的协议数据的数据体，通过解析得到数据体的数据格式、数据主题；

将解析得到的数据格式、数据主题和数据体合并，重新封装成一个消息，实现反序列化。

4.根据权利要求1所述的车辆多源异构数据接入方法，其特征在于：该方法还包括：若由kafka集群向disruptor消息队列返回了消息，则将disruptor消息队列的消息序列化后，通过对应的协议处理器向数据接入请求方返回数据。

5.根据权利要求4所述的车辆多源异构数据接入方法，其特征在于：所述将disruptor消息队列的消息序列化，具体包括：

解析disruptor消息队列的消息，得到消息中包含的数据格式、数据主题和数据体；

将数据体按照对应协议形成协议数据。

6.一种车辆多源异构数据处理方法，其特征在于，该方法对于按照权1-5中任意一项方法接入的数据进行数据处理，该方法具体包括：

接收kafka集群的kafka消息队列发送的消息；

解析接收到的消息，得到消息中包含的数据体以及数据体的数据格式、数据主题；

根据解析得到的数据格式，调用对应的数据格式解析器对数据体进行解析，得到统一格式的数据内容；

将数据内容按照对应数据主题的字段转换配置对应关系，重新封装成统一预设数据结构的数据；

将统一预设数据结构的数据投递到disruptor消息队列。

7.一种车辆多源异构数据接入模块，其特征在于包括：

协议判断单元，用于当有车端、路侧端或第三方平台发送数据接入请求时，判断数据接入请求所采用的协议；

协议数据读取单元，用于调用所采用的协议对应的协议处理器读取当前协议数据；

反序列化单元，用于将接收的协议数据进行反序列化；

disruptor消息队列，用于暂存反序列化的消息；

批量队列消息投递单元，用于批量将disuptor队列中的数据放入kafka集群的kafka消息队列；

以及kafka集群。

8.根据权利要求7所述的车辆多源异构数据接入模块，其特征在于还包括：

序列化单元，用于在kafka集群向disruptor消息队列返回了消息时，将disruptor消息队列的消息序列化，使得序列化后的通过对应的协议处理器向数据接入请求方返回数据。

9.一种车辆多源异构数据处理模块，其特征在于包括：

数据解析单元，用于解析接收到的消息，得到消息中包含的数据体以及数据体的数据格式、数据主题；

数据格式解析单元，用于根据解析得到的数据格式，调用对应的数据格式解析器对数据体进行解析，得到统一格式的数据内容；

数据封装单元，用于将数据内容按照对应数据主题的字段转换配置对应关系，重新封装成统一预设数据结构的数据；

disruptor消息队列，用于暂存统一预设数据结构的数据。

10.一种车辆多源异构数据的中心云处理系统，其特征在于包括：

如权利要求7或8所述的车辆多源异构数据接入模块；

如权利要求9所述的车辆多源异构数据处理模块；以及

数据存储模块，包括存储实时数据的memcache和redis数据库，和存储历史数据的ClickHouse数据库。

Images