CN112527528A - 基于消息队列的数据传输方法、装置及存储介质 - Google Patents

Abstract

本申请公开了一种基于消息队列的数据传输方法、装置、计算机设备及存储介质，该方法包括：接收消息生产者发送的消息数据，消息数据中携带消息生产者生成的消息和消息关联的数据源信息；确定数据源信息关联的消息队列，并根据数据源信息生成目标消息主题；在消息队列中确定出目标消息主题关联的目标存储区域，并将目标消息主题和消息存储至目标存储区域；当接收消息消费者发送的携带数据源信息的消息订阅请求后，根据数据源信息从消息队列的目标存储区域中拉取目标消息主题对应的消息并向消息消费者推送消息。采用本申请，可提高往消息队列中存储数据以及从消息队列中拉取数据的操作便捷性，提高数据传输的效率，适用性高。

Description

基于消息队列的数据传输方法、装置及存储介质

技术领域

本申请涉及计算机技术领域，尤其涉及一种基于消息队列的数据传输方法、装置及存储介质。

背景技术

在大数据领域，消息队列是分布式系统中重要的组件，消息队列这种中间件主要解决应用耦合、异步消息、流量削峰等问题。现有基于消息队列的生产者和消费者应用中，上游生产者应用根据事先定义好的消息主题生成消息并发送到消息队列中间件中，下游消费者应用根据上游生产者应用事先约定好的消息主题消费应对的消息。然而在这样的应用模式中，当上游生产者生成的消息中消息主题发生变更时，必须通知下游消费者应用，开发人员对消费者应用做代码调整和二次开发以适应上游生产者生成的新的消息主题。此外，当消息队列中间件进行重新选型或技术升级时，上游生产者和下游消费者两端的应用架构都需要进行重构和代码的重新开发才能将消息存储至消息队列中或者从消息队列中拉取数据，操作繁琐，实现成本高，适用性差。

发明内容

本申请提供一种基于消息队列的数据传输方法、装置及存储介质，可提高往消息队列中存储数据以及从消息队列中拉取数据的操作便捷性，提高数据传输的效率，适用性高。

第一方面，本申请提供了一种基于消息队列的数据传输方法，该方法适用于消息管理者，该方法包括：

消息管理者接收消息生产者发送的消息数据，上述消息数据中携带上述消息生产者生成的消息和上述消息关联的数据源信息；

上述消息管理者确定上述数据源信息关联的消息队列，并根据上述数据源信息生成目标消息主题；

上述消息管理者在上述消息队列中确定出上述目标消息主题关联的目标存储区域，并将上述目标消息主题和上述消息存储至上述目标存储区域；

当接收到消息消费者发送的消息订阅请求且上述消息订阅请求中携带上述数据源信息时，上述消息管理者根据上述数据源信息从上述消息队列的上述目标存储区域中确定出上述目标消息主题，从上述目标存储区域中拉取上述目标消息主题对应的上述消息并向上述消息消费者推送上述消息。

结合第一方面，在一种可能的实施方式中，上述消息队列中包括上述目标存储区域在内的至少一个存储区域，一个存储区域关联一个消息主题并用于存储所述消息主题的消息，同一个消息主题的消息关联的数据源信息相同。

结合第一方面，在一种可能的实施方式中，上述数据源信息中包括上述消息所属数据源的数据源名称和上述数据源所属项目的项目名称，上述数据源所属项目中包括上述数据源在内的至少一个数据源；

上述根据上述数据源信息生成目标消息主题包括：

根据上述项目名称和上述数据源名称生成目标消息主题，其中，一个数据源名称对应一个消息主题。

结合第一方面，在一种可能的实施方式中，上述消息数据中包括携带在上述消息的头部信息中的消息模式标识，上述消息模式标识用于标识上述消息的消息模式，以供上述消息消费者基于所述消息模式对上述消息进行消费。

结合第一方面，在一种可能的实施方式中，上述消息管理者接收消息生产者发送的消息数据之后，上述方法还包括：

上述消息管理者将所述消息的消息模式注册到消息注册服务中心，通过上述消息注册服务中心生成上述消息的消息模式对应的消息模式标识；

当接收到上述消息消费者发送的消息模式获取请求且上述消息模式获取请求中携带上述消息模式标识时，从上述消息注册服务中心获取上述消息模式标识对应的上述消息模式向所述消息消费者发送上述消息模式。

第二方面，本申请提供了一种基于消息队列的数据传输方法，该方法适用于消息消费者，该方法包括：

消息消费者获取消息生产者创建的数据源的数据源信息；

当上述消息消费者请求订阅上述消息生产者生成的消息时，向消息管理者发送消息订阅请求，上述消息订阅请求中携带数据源信息，上述数据源信息用于确定上述消息消费者请求订阅的目标消息主题；

上述消息消费者从上述消息管理者接收上述目标消息主题关联的消息，解析并消费上述消息。

结合第二方面，在一种可能的实施方式中，上述解析并消费所述消息，包括：

上述消息消费者从上述消息的头部信息中获取消息模式标识信息，并根据上述消息模式标识信息确定上述消息的消息模式；

基于上述消息的消息模式对上述消息进行解析后消费。

结合第二方面，在一种可能的实施方式中，上述解析并消费上述消息，还包括：

上述消息消费者从上述消息的头部信息中获取消息模式标识，并向消息管理者发送消息模式获取请求，上述消息模式获取请求中携带上述消息模式标识；

从上述消息管理者获取上述消息模式标识对应的消息模式，并基于上述消息模式对上述消息进行解析后消费。

第三方面，一种基于消息队列的数据传输方法，该方法适用于消息生产者，该方法包括：

创建数据源，并根据上述数据源生成消息和上述消息关联的数据源信息；

向消息管理者发送消息数据，上述消息数据中携带所述消息和所述数据源信息。

结合第三方面，在一种可能的实施方式中，上述消息生产者创建数据源之后，该方法还包括：

向消息管理者和/或消息消费者发送上述数据源的数据源信息；

其中，上述数据源信息包括数据源名称和数据源所属项目的项目名称。

第四方面，本申请提供了一种基于消息队列的数据传输装置，上述数据传输装置适用于消息管理者，上述数据传输装置包括：

接收模块，用于接收消息生产者发送的消息数据，上述消息数据中携带上述消息生产者生成的消息和上述消息关联的数据源信息；

确定模块，用于确定上述消息数据接收模块接收的上述数据源信息关联的消息队列，并根据上述数据源信息生成目标消息主题；

上述确定模块，还用于在上述消息队列中确定出上述目标消息主题关联的目标存储区域，并将上述目标消息主题和上述消息存储至上述目标存储区域；

上述接收模块，还用于接收消息消费者发送的消息订阅请求，上述消息订阅请求中携带上述数据源信息；

消息推送模块，用于根据上述接收模块接收的消息订阅请求中携带的上述数据源信息从上述消息队列的上述目标存储区域中确定出上述目标消息主题，从上述目标存储区域中拉取上述目标消息主题对应的上述消息并向上述消息消费者推送上述消息。

结合第四方面，在一种可能的实施方式中，上述消息队列中包括上述目标存储区域在内的至少一个存储区域，一个存储区域关联一个消息主题并用于存储上述消息主题的消息，同一个消息主题的消息关联的数据源信息相同。

结合第四方面，在一种可能的实施方式中，上述数据源信息中包括上述消息所属数据源的数据源名称和上述数据源所属项目的项目名称，上述数据源所属项目中包括上述数据源在内的至少一个数据源；上述确定模块用于：

根据上述项目名称和上述数据源名称生成目标消息主题，其中，一个数据源名称对应一个消息主题。

结合第四方面，在一种可能的实施方式中，上述基于消息队列的消息传输装置还包括：

注册模块，用于将上述消息模式注册到消息注册服务中心，并通过上述消息注册服务中心生成上述消息的消息模式对应的消息模式标识；

上述消息推送模块还用于：

当接收到上述消息消费者发送的消息模式获取请求且上述消息模式获取请求中携带上述消息模式标识后，从上述消息注册服务中心获取上述消息模式标识对应的上述消息模式向上述消息消费者发送上述消息模式。

第五方面，本申请提供了一种基于消息队列的数据传输装置，上述数据传输装置适用于消息消费者，所述数据传输装置包括：

获取模块，用于获取消息生产者创建的数据源的数据源信息；

发送模块，用于在订阅上述消息生产者生成的消息时，向消息管理者发送消息订阅请求，上述消息订阅请求中携带数据源信息，上述数据源信息用于确定上述请求订阅的目标消息主题；

接收模块，用于从上述消息管理者接收上述目标消息主题关联的消息；

解析消费模块，用于解析并消费上述接收模块接收的上述消息。

结合第五方面，在一种可能的实施方式中，上述解析消费模块用于：

从上述接收模块接收的消息的头部信息中获取消息模式标识信息，并根据上述消息模式标识信息确定上述消息的消息模式；

基于上述消息的消息模式对上述消息进行解析后消费。

结合第五方面，在一种可能的实施方式中，上述解析消费模块用于：

从上述接收模块接收的消息的头部信息中获取消息模式标识，并通过上述发送模块向消息管理者发送消息模式获取请求，上述消息模式获取请求中携带上述消息模式标识；

从上述消息管理者获取上述消息模式标识对应的消息模式，并基于上述消息模式对上述消息进行解析后消费。

第六方面，本申请提供了一种基于消息队列的数据传输装置，上述数据传输装置适用于消息生产者，所述数据传输装置包括：

创建模块，用于创建数据源；

消息生成模块，根据上述创建模块创建的上述数据源生成消息和上述消息关联的数据源信息；

发送模块，用于向消息管理者发送消息数据，上述消息数据中携带上述消息生成模块生成的上述消息和上述数据源信息。

结合第六方面，在一种可能的实施方式中，上述发送模块还用于：

向消息管理者和/或消息消费者发送上述数据源的数据源信息；

其中，上述数据源信息包括数据源名称和数据源所属项目的项目名称。

第七方面，本申请提供了一种计算机设备，包括：处理器、存储器、收发器、网络接口；

该处理器与存储器、网络接口相连，其中，网络接口用于提供数据通信功能，该存储器用于存储计算机程序，该处理器和收发器用于调用该计算机程序，以执行本申请中上述第一方面以及第一方面中任一种可能的实施方式所执行的方法。

第八方面，本申请提供了一种计算机设备，包括：处理器、存储器、收发器、网络接口；

该处理器与存储器、网络接口相连，其中，网络接口用于提供数据通信功能，该存储器用于存储计算机程序，该处理器和收发器用于调用该计算机程序，以执行本申请中上述第二方面以及第二方面中任一种可能的实施方式所执行的方法。

第九方面，本申请提供了一种计算机设备，包括：处理器、存储器、收发器、网络接口；

该处理器与存储器、网络接口相连，其中，网络接口用于提供数据通信功能，该存储器用于存储计算机程序，该处理器和收发器用于调用该计算机程序，以执行本申请中上述第三方面以及第三方面中任一种可能的实施方式所执行的方法。

第十方面，本申请提供了一种计算机可读存储介质，该计算机可读存储介质存储有计算机程序，该计算机程序包括程序指令，该程序指令当被处理器执行时，执行本申请中上述第一方面以及第一方面中任一种可能的实施方式所执行的方法。

第十一方面，本申请提供了一种计算机可读存储介质，该计算机可读存储介质存储有计算机程序，该计算机程序包括程序指令，该程序指令当被处理器执行时，执行本申请中上述第二方面以及第二方面中任一种可能的实施方式所执行的方法。

第十二方面，本申请提供了一种计算机可读存储介质，该计算机可读存储介质存储有计算机程序，该计算机程序包括程序指令，该程序指令当被处理器执行时，执行本申请中上述第三方面以及第三方面中任一种可能的实施方式所执行的方法。

在本申请中，消息生产者生成消息时可以自行创建数据源，当数据源创建后，消息管理者会操作绑定对应的消息队列组件，并在对应的消息队列中创建一个与数据源绑定对应关系的目标存储区域，消息消费者以及消息生产者可以重复使用此数据源，消息管理者可以往数据源对应的目标存储区域存储数据或者拉取数据，可提高消息管理者往消息队列中存储数据以及从消息队列中拉取数据的操作便捷性，提高数据传输的效率，适用性高。此时消息队列对于消息生产者和消息消费者都是透明的，消息生产者不需要关心具体使用的消息队列这种中间件是什么组件，他们之间是通过数据源来关联的，只要消息生产者绑定了某个数据源，后续如果消息生产者更改底层消息队列的话，消息管理者在消息队列的组件类型切换时，根据消息生产者绑定的数据源，在新的消息队列中创建新的目标存储区域，消息生产者的应用可以进行无缝切换，不需要修改应用端的程序和配置，可减少消息生产者或者消息消费者学习不同类型的消息队列的成本，简化了消息队列应用的开发模式，降低开发技术难度，帮助业务方更快，更高效的实业上层应用的业务逻辑。

附图说明

为了更清楚地说明本申请中的技术方案，下面将对实施例描述中所需要使用的附图作简单地介绍，显而易见地，下面描述中的附图仅仅是本申请的一些实施例，对于本领域普通技术人员来讲，在不付出创造性劳动的前提下，还可以根据这些附图获得其他的附图。

图1是本申请提供的基于消息队列的数据传输系统架构的结构示意图；

图2是本申请提供的基于消息队列的数据传输的交互示意图；

图3是本申请提供的基于消息队列的数据传输方法的一流程示意图；

图4是本申请提供的基于消息队列的数据传输方法的另一流程示意图；

图5是本申请提供的基于消息队列的数据传输方法的另一流程示意图；

图6是本申请提供的基于消息队列的数据传输装置的结构示意图；

图7是本申请提供的基于消息队列的数据传输装置的另一结构示意图；

图8是本申请提供的基于消息队列的数据传输装置的另一结构示意图；

图9是本申请提供的计算机设备的结构示意图；

图10是本申请提供的计算机设备的另一结构示意图；

图11是本申请提供的计算机设备的另一结构示意图。

具体实施方式

下面将结合本申请中的附图，对本申请中的技术方案进行清楚、完整地描述，显然，所描述的实施例仅仅是本申请一部分实施例，而不是全部的实施例。基于本申请中的实施例，本领域普通技术人员在没有做出创造性劳动前提下所获得的所有其他实施例，都属于本申请保护的范围。

请参见图1，图1是本申请提供的一种基于消息队列的数据传输系统架构的结构示意图。如图1所示，该网络架构可以包括服务器20、用户终端集群，该用户终端集群可以包括多个用户终端，如图1所示，具体可以包括用户终端100a、用户终端100b、用户终端100c、…、用户终端100n。

可以理解的是，在本申请可选的实施例中消息生产者、消息消费者可以是上述用户终端集群中的用户终端，消息生产者可以是生成并往消息的队列中存储消息的用户终端，消息消费者可以是订阅并消费消息队列中存储的消息的用户终端，为方便描述，下面将以消息生产者和消息消费者为例进行说明。上述用户终端可以包括但不限于手机、平板电脑、笔记本电脑、掌上电脑、智能音响、移动互联网设备(mobile internet device，MID)、销售点(point of sales，POS)、可穿戴设备(例如智能手表、智能手环等)。在本申请实施例中，上述消息管理者、消息队列、数据源、消息注册服务中心可以为同一个服务器的一部分，也可以分布在不同的服务器中，上述服务器可以是独立的物理服务器，也可以是多个物理服务器构成的服务器集群或者分布式系统，还可以是提供云服务、云数据库、云计算、云函数、云存储、网络服务、云通信、中间件服务、域名服务、安全服务、CDN、以及大数据和人工智能平台等基础云计算服务的云服务器，具体可根据实际应用场景确定，在此不做限制。为方便描述，下面将以消息管理者作为分布式系统中的服务器或者消息传输中心(如图1中的服务器20)为例进行说明。消息队列可以是分布式系统中的中间件，用于存储消息生产者生成的消息。消息管理者可以从消息生产者接收消息并存储至消息队列中，也可以从消息队列中拉取数据并推送给消息消费者，还可以将消息的消息类型等消息模式注册到消息注册服务中心，或者从消息注册服务中心获取消息消费者请求的消息模式并下发给消息消费者等。

在一些可行的实施方式中，如图1所示，用户终端100a、用户终端100b、用户终端100c、…、用户终端100n可以分别与上述服务器20进行网络连接，以便于每个用户终端可以通过该网络连接与服务器20进行数据交互。例如，用户终端集群中的任意一个装载有消息生产应用的用户终端(即消息生产者，如用户终端100a)向服务器20(即消息管理者)发送携带有数据源信息的消息数据，也可以是，服务器20在接收到上述用户终端集群中的任意一个装载有消息消费应用的用户终端(即消息消费者，如用户终端100b)发送的携带有数据源信息的消息订阅请求时，向该用户终端(如用户终端100b)发送对应的消息数据。其中上述消息生产应用以及消息消费应用可以是社交应用或者网购应用等，上述消息数据可以是用户发送的聊天消息也可以是订单消息，具体可根据实际用于场景确定，在此不做限制。

请参见图2，图2是本申请提供的一种基于消息队列的数据传输的交互示意图。如图2所示，本申请提供的基于消息队列的数据传输系统可以包括消息生产者10a、消息管理者10b、消息消费者10c、数据源10d、消息注册中心10e、消息队列10f。其中，图1中的数据源10d为消息生产者10a或消息消费者10c创建且两者可以共同使用的数据源，或者协议规定的数据源，比如消息生产者10a绑定的数据源，之后通过协议的方式告知消息消费者10c。这里的数据源10d可以包括关系型数据库(如Mysql、Oracle、Nosql数据库)还可以包括“消息缓存”这种数据源。“消息缓存”是指可以重复利用的数据源，它是一种逻辑概念。这里，数据源名字可以由用户自定义，数据源名称可以由英文全小写字母、下划线或者数字组成。创建数据源10d还需要项目，项目是用于区分用户所创建的不同用途的数据源的集合项目是实时流计算系统的基础部分，在进行其他操作之前，必须先创建项目，然后可在项目管理中对各个项目下的数据源，实时计算，资源和成员进行管理。数据源10d的源端主要对接数据库(Mysql/PG/SQLServer/Oracle)\采集日志文件、JQData、Kafka等多种类型的数据库。数据源10d用于与消息列队10f进行对接，将消息生产者10a和消息消费者10c关联在一起，再从消息队列10f中将数据推送到实时流计算系统，然后用于下游计算任务。实时计算主要指与消息列队10f对接的实时计算任务，如flink或spark streaming任务。当消息生产者10a或消息消费者10c创建数据源10d后上报给消息管理者10b，消息管理者10b绑定与该数据源10d对应的消息队列(如消息队列10f)，并根据上述数据源信息生成与之对应的目标消息主题。其中，数据源10d与消息主题之间的对应关系可以为，消息主题＝项目名称+数据源名称。可以理解，通过数据源10d可以在消息队列10f中直接找到对应的消息主题，并通过消息主题和消息的对应关系将消息和数据源10d关联起来，进而可实现通过数据源10d找到消息队列10f中的消息主题，通过消息主题找到消息。因此消息管理者10b可以通过数据源10d找到消息队列10f中的消息主题，通过消息主题找到消息，从而实现在对应的消息队列10f中对应的目标存储区域进行数据的生产和消费等操作。此时所创建的数据源10d可以是作为source输入端，也可作为sink输出端，作为flink等各种形式的实时计算引擎的数据来源，也可以是实时计算之后数据流向的地方。

其中，图2中的消息生产者10a为产生消息数据的用户终端。消息生产者10a在生产消息数据时可先自行创建数据源10d，消息管理者10b会根据消息生产者10a创建的数据源10d操作绑定对应的消息队列10f，并自动在对应的消息队列10b中创建同样与数据源10d对应的目标存储区域。消息生产者10a将生成的消息数据发送至消息管理者10b，由消息管理者10b将上述接收的消息数据存放至消息队列10f中的目标存储区域。只要消息生产者10a创建了某个数据源10d，后续如果消息队列10b的中间件类型进行切换的时候，消息管理者10b会根据消息生产者10a创建的数据源10d，在新的消息队列中创建一个新的目标存储区域，用于存储根据上述消息生产者10a发送至消息管理者10b的消息数据。这样消息生产者10a的应用可以进行无缝切换，不需要修改应用端的程序和配置。

其中，图2中的消息队列10f是分布式系统中的重要组件，主要解决应用耦合、异步消息、流量削峰等问题。在本申请实施例中，消息队列10f相当于保存消息的一个容器。它的特点在于，经由消息队列10f的消息数据只能先进再出，消息管理者10b先通过数据源10d这个通用的逻辑概念将携带有消息生产者10a生成的消息以及根据数据源信息生成的目标消息主题的消息数据投递给消息队列10f中的目标存储区域保存。消息消费者10c会监听消息队列10f中的数据，如果消息队列10f中有数据可供消费，再同样使用上述数据源10d这个通用的逻辑概念向消息管理者10b发送消息订阅请求，消息管理者10b会根据上述数据源信息从消息队列10f中的目标存储区域中确定出消息订阅请求的目标消息主题，从上述目标存储区域中拉取目标消息主题对应的消息，并将上述消息推送给消息消费者10c逐条进行消费。因为消息队列10f接收数据的性能是非常高的，不会影响消息生产者10a的性能，也保证了消息消费者10c的及时效率。在本申请可选的实施例中，消息队列10f对于消息生产者10a以及消息消费者10b是透明的，他们之间通过数据源10d来相互关联，上述消息队列10f可以为activeMQ、rabbitMQ、rocketMQ、zeroMQ、kafka等消息队列中间件。

其中，图2中的消息消费者10c为处理数据的用户终端。消息消费者10c通过与消息生产者10a共用的数据源，向消息管理者10b发送携带数据源信息的消息订阅请求。消息管理者10b根据上述数据源信息从消息队列10f的目标存储区域确定出上述消息订阅请求订阅的消息主题，从上述目标存储区域中拉取出目标消息主题对应的消息，并将上述消息推送给消息消费者10c。消息消费者10c接收返回的订阅消息后，解析出消息头部信息中的消息模式标识信息，上述消息模式标识信息用于标识消息的消息模式，消息消费者10c根据消息模式使用相应的解析方法解析并消费消息。

在一些可行的实施例中，若消息头部信息中没有消息模式标识信息，上述消息消费者10c还可以向消息管理者10b发送携带消息模式标识的消息模式获取请求。消息管理者10b接收到上述请求后从消息注册服务中心10e获取上述消息模式标识对应的消息模式，并将上述消息模式发送给消息消费者10c。消息消费者10c通过上述消息模式解析并消费消息。在本申请可选实施例中，假如消息消费者10c需要订阅某个数据源里面的数据，消息消费者可以引入SDK包调用数据源订阅接口，订阅服务提供了服务端保存用户消费点的功能，只需要通过简单配置和处理，就可以实现高可用的点位存储服务。

其中，在本申请可选的实施例中，图2中的基于消息队列的数据传输系统还可以包括消息注册服务中心10e，上述消息注册服务中心10e用于记录消息管理者10b注册的消息模式。在本申请的可选实施例中，消息管理者10b接收消息生产者10a发送的消息数据后，消息管理者10b将上述消息的消息模式注册到消息注册服务中心10e，并通过消息注册服务中心10e生成消息的消息模式对应的消息模式标识，将上述消息模式标识信息添加至消息的头部消息中随消息数据一起推送。当消息管理者10b接收到消息消费者10c发送的消息模式获取请求且上述消息模式请求携带有上述消息模式标识时，消息管理者10b从消息注册服务中心10e中获取上述消息模式标识对应的消息模式向消息消费者10c发送上述消息模式。可选的，上述消息注册服务中心10e可以为Spring Cloud组件，它是一系列框架的有序集合，包含了微服务构架的方方面面，对各个组件进行了大量的封装，从而简化了开发。

下面将结合图3-图5对本申请实施例提供的基于消息队列的数据传输方法进行示例说明。图3是本申请实施例提供的一种基于消息队列的数据传输方法的一流程示意图。如图3所示，本申请实施例提供的基于消息队列的数据传输方法包括如下步骤：

S101,发送携带数据源信息的消息数据。

在一些可行的实施方式中，消息管理者接收消息生产者发送的消息数据，上述消息数据中携带上述消息生产者生成的消息和上述消息关联的数据源信息。消息生产者在发送上述消息数据之前先创建数据源，将上述创建的数据源关联至某一个项目，并根据上述数据源生成消息和上述消息关联的数据源信息，创建数据源之后将上述携带有消息和数据源信息的消息数据发送至消息管理者，并将上述数据源信息发送至消息消费者。消息消费者将接收到的数据源信息添加至消息订阅请求中一起发送至消息管理者。消息管理者根据数据源从消息队列中存储和拉取消息。其中，上述数据源信息包括数据源名称和数据源类型。这里，数据源名称可以为英文全小写字母、下划线、数字组成，例如“source_123”。可选的，上述创建的数据源可以为“消息缓存”这种数据源，它是一种逻辑概念，指可以重复利用的数据源。这里的项目为区分用户所创建的不同用途的数据源集合(包括一个或者多个数据源)，换句话说，一个或者多个数据源可以关联同一个项目，然后可在项目管理中对各个项目下的数据源、实时计算、资源和成员进行管理。可选的，上述消息数据中还可以包括携带在消息的头部信息中的消息模式标识，上述消息模式标识用于标识上述消息的消息模式，这里的消息模式用于消息消费者基于上述消息模式对消息进行解析后消费。可以理解，上述数据源用于建立消息生产者生成的消息与消息主题以及数据源三者之间的联系，同时上述数据源还用于与消息队列进行对接，将消息生产者与消息消费者关联在一起，再从消息队列中将数据推送到实时流计算系统以用于下游计算任务。数据源创建后将上述数据源信息添加至消息数据中随消息的头部信息一起发送给消息管理者。可选的，在消息传输时可以使用Apache Avro格式传输消息，同时可以使用SDK的方式统一封装管理该功能。可选的，上述消息生产者发送携带数据源信息的消息数据，可以是在终端的某种购物应用中，用户下单后订单进程完成持久化处理将上述用户的下单信息发送至服务器中的消息管理者。

S102，注册上述消息数据的消息模式。

在一些可行的实施方式中，在消息管理者接收消息生产者发送的消息数据之后，消息管理者将上述消息的消息模式到消息注册服务中心，并通过消息注册服务中心生成消息的消息模式对应的消息模式标识，将上述消息模式标识添加至消息的头部消息中随消息数据一起推送。当消息消费者向消息管理者发送消息模式获取请求且上述消息模式请求携带有上述消息模式标识时，消息管理者可从消息注册服务中心中获取上述消息模式标识对应的消息模式向消息消费者发送上述消息模式。在本申请可选的实施例中，上述消息管理者将上述消息数据的消息模式注册至消息注册服务中心可以为，将上述消息数据注册至spring cloud注册中心，并通过spring cloud生成消息的消息模式对应的消息模式标识。

S103，根据数据源信息生成目标消息主题，并将上述目标消息主题对应的消息存储至消息队列中对应的目标存储区域。

在一些可行的实施方式中，当消息管理者接收到上述消息数据后会根据上述数据源信息绑定对应的消息队列，并根据上述数据源信息生成对应的目标消息主题，上述数据源与目标消息主题为一一对应关系。可选的，目标消息主题名称为项目名称加上数据源名称，例如“program1+source123”。可以理解，通过上述数据源就可以在消息队列中找到对应的目标消息主题，并通过目标消息主题和消息的对应关系将消息和数据源关联起来，进而可以实现通过数据源找到消息队列中的目标消息主题，通过目标消息主题找到消息，因此消息生产者与消息消费者可以通过数据源找到底层消息队列中的目标消息主题，通过目标消息主题找到消息，从而实现在消息队列中对应的目标存储区域进行数据的生产和消费操作。上述消息队列中包括上述目标存储区域在内的至少一个存储区域，一个存储区域关联一个消息主题，同一个消息主题的消息关联的数据源信息也相同，上述存储区域用于存储上述消息主题的消息。可选的，上述消息数据进入消息队列中的对应存储区域时可以选择CSV、JSON、AVRO等数据格式，而不需要单独再去设置消息主题的数据字典。当消息生产者或消息消费者创建数据源后，消息管理者会操作绑定对应的消息队列，并在对应的消息队列中操作绑定一个同样与数据源绑定对应关系的目标存储区域用于存储与上述数据源所对应的消息主题的消息。消息消费者和消息生产者可以重复使用这个数据源，也可以往上述数据源对应的目标存储区域中生产推送数据或者消费拉取数据。后续如果消息队列类型进行切换时，消息管理者会根据消息生产者绑定的数据源自动在新的消息队列中创建绑定新的目标存储区域。这样消息生产者的应用可以进行无缝切换，此时消息队列这种中间件对于消息生产者和消息消费者都是透明的，消息生产者与消息消费者不需要关心具体使用的是何种消息队列，就可以进行数据的推送与拉取。

S104，发送携带数据源信息的消息订阅请求。

在一些可行的实施方式中，消息消费者向消息管理者发送携带数据源信息的订阅请求，由于消息消费者与上述消息生产者共用上述数据源，因此当消息管理者接收到消息消费者发送的携带有上述数据源信息的消息订阅请求后，会首先解析上述消息订阅请求，并根据上述消息订阅请求所携带的数据源信息找到与上述数据源在消息队列中相对应的目标存储区域所存储的对应的消息主题的消息。上述数据源信息包括数据源名称，根据上述数据源名称找到与上述数据源名称相对应的消息主题的名称，以确定消息订阅请求所订阅的消息主题。在本申请可选的实施例中，上述消息消费者请求订阅消息生产者生成的消息，可以为终端中的购物应用中的库存管理进程，向服务器中的消息管理者通过约定的数据源发送用户的下单信息订阅请求，当库存管理进程订阅用户的下单信息，采用拉取的方式，获取用户下单信息，并根据上述用户下单信息，进行对应的库存操作。

S105，返回根据上述数据源信息拉取的对应存储区域中的消息。

在一些可行的实施方式中，消息管理者根据上述数据源信息中的数据源名称找到与上述数据源名称相对应的消息主题的名称后，从存储上述消息主题的消息的目标存储区域中拉取消息消费者所订阅的消息，并将上述消息推送给消息消费者。

S106，从接收的消息的头部信息中获取消息模式标识。

在一些可行的实施方式中，当消息消费者接收到来自消息管理者发送的存储在消息队列中的目标存储区域的消息后，针对不同的数据格式采用对应的格式解析方法，从消息的头部信息中解析出消息模式标识，消息消费者根据获取的消息模式标识确定出上述根据消息订阅请求接收到的消息的消息模式，这里的消息模式用于消息消费者解析消息。如果消息生产者生成的消息对应的消息模式发生变更，消息消费者直接根据消息的头部信息获取到变更后的消息模式标识。可选的，若上述消息消费者接收的消息头部信息中没有消息模式标识信息，则消息消费者可执行步骤S107。

S107，发送携带消息模式标识的消息模式请求。

在一些可行的实施方式中，消息消费者根据从消息的头部信息中解析出的消息模式标识向消息管理者发送携带消息模式标识的消息模式请求。

S108，从消息注册服务中心拿取消息模式标识对应的消息模式。

在一些可行的实施方式中，消息管理者根据接收到的携带消息模式标识在消息注册服务中心拿取对应的消息模式，用以发往消息消费者。

S109，发送从消息注册服务中心获取的对应消息模式。

在一些可行的实施方式中，当消息消费者向消息管理者发送消息模式获取请求且上述消息模式请求携带有上述消息模式标识时，消息管理者从消息注册服务中心中获取上述消息模式标识对应的消息模式向消息消费者发送上述消息模式。

S110，基于获取的消息模式对上述消息进行解析后消费。

在一些可行的实施方式中，消息消费者根据不同的消息的消息模式对消息使用对应的解析方法进行解析后消费。上述消息模式包括但不限于消息的格式类型以及消息队列的类型，由于消息消费者与消息生产者共用数据源这个通用的逻辑概念，避免了对消息主题数据字典的管理，只需对接收的消息的消息模式根据对应的解析方法进行解析后下游的消息消费者就可以对消息进行消费。

在本申请中，消息生产者或消息消费者自行创建数据源，由消息管理者统一维护数据源与底层消息队列中消息主题的映射关系、消息队列对消息生产者及消费者之间的逻辑关系，呈现给消息消费者和消息生产者应用侧的只有抽象后的总线数据流转模型，消息消费者和消息生产者仅需了解数据源名称和目标存储名称就可以进行消息的生成和消费等。这里，数据源用于与消息队列进行对接，将消息生产者与消息消费者关联在一起，再从消息队列中将数据推送到实时流计算系统，然后用于后续计算任务。消息消费者可以根据与消息生产者共用的数据源信息识别数据源对应的消息主题，不需要对消息主题数据字典进行管理，并能根据不同消息的消息模式使用相应的解析方法解析消息，操作简单适用性高。

请参见图4，图4是本申请提供的一种基于消息队列的数据传输方法的另一流程示意图。该方法适用于消息管理者侧，该方法包括如下步骤：

S201，消息管理者接收消息生产者发送的消息数据，上述消息数据中携带上述消息生产者生成的消息和上述消息关联的数据源信息。

其中，该步骤S201的具体实施方式可参见上述图3所对应实施例中对步骤S101的描述，这里将不再赘述。

S202，消息管理者确定上述数据源信息关联的消息队列，并根据上述数据源信息生成目标消息主题。

其中，该步骤S202的具体实施方式可参见上述图3所对应实施例中对步骤S103的描述，这里将不再赘述。

S203，消息管理者在上述消息队列中确定出上述目标消息主题关联的目标存储区域，并将上述目标消息主题和上述消息存储至上述目标存储区域。

其中，该步骤S203的具体实施方式可参见上述图3所对应实施例中对步骤S103的描述，这里将不再赘述。

S204，当接收到消息消费者发送的消息订阅请求且上述消息订阅请求中携带上述数据源信息时，上述消息管理者根据上述数据源信息从上述消息队列的上述目标存储区域中确定出上述目标消息主题。

在一些可行的实施方式中，消息消费者与上述消息生产者共用上述数据源，因此当消息管理者接收到消息消费者发送的携带有上述数据源信息的消息订阅请求后，会首先解析上述消息订阅请求，并根据上述消息订阅请求所携带的数据源信息找到与上述数据源在消息队列中相对应的目标存储区域所存储的对应的消息主题的消息。上述数据源信息包括数据源名称，根据上述数据源名称找到与上述数据源名称相对应的消息主题的名称，以确定消息订阅请求所订阅的消息主题。

S205，从上述目标存储区域中拉取上述目标消息主题对应的上述消息并向上述消息消费者推送上述消息。

在一些可行的实施方式中，消息管理者根据上述数据源信息中的数据源名称找到与上述数据源名称相对应的消息主题的名称后，从存储上述消息主题的消息的目标存储区域中拉取消息消费者所订阅的消息，并将上述消息推送给消息消费者。

在本申请中，消息生产者生成消息时可以自行创建数据源当数据源创建后，消息管理者会操作绑定对应的消息队列组件，并在对应的消息队列中创建一个与数据源绑定对应关系的目标存储区域，只要消息生产者绑定了某个数据源，后续如果消息生产者更改底层消息队列的话，消息管理者在消息队列的组件类型切换时，根据消息生产者绑定的数据源，在新的消息队列中创建新的目标存储区域，消息生产者的应用可以进行无缝切换，不需要修改应用端的程序和配置。可简化消息队列应用的开发模式，提高消息管理者往消息队列中存储数据以及从消息队列中拉取数据的操作便捷性。

请参见图5，图5是本申请提供的一种基于消息队列的数据传输方法的一流程示意图。该方法适用于消息消费者侧，该方法包括如下步骤：

S301，消息消费者获取消息生产者创建的数据源的数据源信息。

在一些可行的实施方式中，上述由消息生产者创建的数据源可以为消息消费者所共用，可选的，上述数据源也可以是由消息消费者创建供消息生产者共用的。上述数据源信息包括但不限于数据源名称和数据源类型。在本申请可选的实施例中，此时所创建的数据源不仅是作为source输入端，也可作为sink输出端，作为flink等各种形似的实时计算引擎的数据来源，也可以是实时计算之后数据流向的地方，这里的实时计算主要指与消息队列组件对接的实时计算任务，如flink或spark streaming任务。

S302，当上述消息消费者请求订阅上述消息生产者生成的消息时，向消息管理者发送消息订阅请求，上述消息订阅请求中携带数据源信息，上述数据源信息用于确定上述消息消费者请求订阅的目标消息主题。

在一些可行的实施方式中，当消息消费者需要订阅消息时，将上述获取的用于确定目标消息主题的数据源信息随上述消息订阅请求一起发送至消息管理者。这里的数据源信息与目标消息主题的对应关系具体请参见步骤S103，在此则不再过多说明。

S303，消息消费者从上述消息的头部信息中获取消息模式标识信息，并根据上述消息模式标识信息确定上述消息的消息模式。

在一些可行的实施方式中，当消息消费者接收到来自消息管理者发送的存储在消息队列中的目标存储区域的消息后，针对不同的数据格式采用对应的格式解析方法，从消息的头部信息中解析出消息模式标识信息，上述消息模式标识信息包括但不限于消息模式标识，消息模式等。消息消费者根据获取的消息标识模式信息确定出上述根据消息订阅请求接收到的消息属于何种消息模式。如果消息生产者生成的消息对应的消息模式发生变更，消息消费者直接根据消息的头部信息获取到变更后的消息模式，并根据接收到的消息模式解析并消费变更之后的消息。

S304，基于上述消息的消息模式对上述消息进行解析后消费。

其中，该步骤S304的具体实施方式可参见上述图3所对应实施例中对步骤S109的描述，这里将不再赘述。

在本申请中，消息生产者或消息消费者自行创建数据源，由消息管理者统一维护数据源与底层消息队列中消息主题的映射关系、消息队列对消息生产者及消息消费者之间的逻辑关系，呈现给消息消费者侧的只有抽象后的总线数据流转模型，消息消费者和消息生产者仅需了解数据源名称和目标存储名称就可以进行消息的生成和消费等。这里，数据源用于与消息队列进行对接，将消息生产者与消息消费者关联在一起，再从消息队列中将数据推送到实时流计算系统，然后用于后续计算任务。消息消费者可以根据与消息生产者共用的数据源信息识别数据源对应的消息主题，不需要对消息主题数据字典进行管理，并能根据不同消息的消息模式使用相应的解析方法解析消息，操作简单适用性高。

进一步地，请参见图6，图6是本申请提供的基于消息队列的数据传输装置的一结构示意图。该基于消息队列的数据传输装置可以是对应于消息管理者的用户终端，也可以是运行于计算机设备中的一个计算机程序(包括程序代码)，例如，该基于消息队列的数据传输装置为一个应用软件；该基于消息队列的数据传输装置可以用于执行本申请提供的方法中的相应步骤。如图6所示，该基于消息队列的数据传输装置1可以应用于计算机设备，基于消息队列的数据传输装置1可以包括：接收模块11、确定模块12、消息推送模块13、注册模块14。

接收模块11，用于接收消息生产者发送的消息数据，上述消息数据中携带上述消息生产者生成的消息和上述消息关联的数据源信息；

确定模块12，用于确定上述消息数据接收模块接收的上述数据源信息关联的消息队列，并根据上述数据源信息生成目标消息主题；

上述确定模块12，还用于在上述消息队列中确定出上述目标消息主题关联的目标存储区域，并将上述目标消息主题和上述消息存储至上述目标存储区域；

上述接收模块11，还用于接收消息消费者发送的消息订阅请求，上述消息订阅请求中携带上述数据源信息；

消息推送模块13，用于根据上述接收模块接收的消息订阅请求中携带的上述数据源信息从上述消息队列的上述目标存储区域中确定出上述目标消息主题，从上述目标存储区域中拉取上述目标消息主题对应的上述消息并向上述消息消费者推送上述消息。

在一种可能的实施方式中，上述消息队列中包括上述目标存储区域在内的至少一个存储区域，一个存储区域关联一个消息主题并用于存储上述消息主题的消息，同一个消息主题的消息关联的数据源信息相同。

在一种可能的实施方式中，所述数据源信息中包括所述消息所属数据源的数据源名称和所述数据源所属项目的项目名称，所述数据源所属项目中包括所述数据源在内的至少一个数据源；上述确定模块用于：

根据所述项目名称和所述数据源名称生成目标消息主题，其中，一个数据源名称对应一个消息主题。

在一种可能的实施方式中，上述基于消息队列的消息传输装置还包括：

注册模块14，用于将上述消息模式注册到消息注册服务中心，并通过上述消息注册服务中心生成上述消息的消息模式对应的消息模式标识。

在一种可能的实施方式中，上述消息推送模块13还用于：

当接收到上述消息消费者发送的消息模式获取请求且上述消息模式获取请求中携带上述消息模式标识后，从上述消息注册服务中心获取上述消息模式标识对应的上述消息模式向上述消息消费者发送上述消息模式。

其中，接收模块11、确定模块12、推送模块13、注册模块14的具体功能的实现方式可以参见上述图3对应实施例中的步骤S101-S105，这里不再进行赘述。

在本申请中，消息生产者生成消息时可以自行创建数据源当数据源创建后，消息管理者会操作绑定对应的消息队列组件，并在对应的消息队列中创建一个与数据源绑定对应关系的目标存储区域，消息消费者以及消息生产者可以重复使用此数据源，也可以往数据源对应的目标存储区域生产消费数据或者消费拉取数据。此时消息队列对于消息生产者和消息消费者都是透明的，消息生产者不需要关心具体使用的消息队列这种中间件是什么组件，他们之间是通过数据源来关联的，只要消息生产者绑定了某个数据源，后续如果消息生产者更改底层消息队列的话，消息管理者在消息队列的组件类型切换时，根据消息生产者绑定的数据源，在新的消息队列中创建新的目标存储区域，消息生产者的应用可以进行无缝切换，不需要修改应用端的程序和配置。可减少消息生产者或者消息消费者学习不同类型的消息队列的成本，提高往消息队列中存储数据以及从消息队列中拉取数据的操作便捷性，提高数据传输的效率。

请一并参见图7，是本发明实施例提供的另一种基于消息队列的数据传输装置的结构示意图。如图7所示，基于消息队列的数据传输装置2可以是对应于消息消费者的用户终端，该数据传输装置2包括：获取模块21、发送模块22、接收模块23、解析消费模块24。

获取模块21，用于获取消息生产者创建的数据源的数据源信息；

发送模块22，用于订阅上述消息生产者生成的消息时，向消息管理者发送消息订阅请求，上述消息订阅请求中携带数据源信息，上述数据源信息用于确定上述请求订阅的目标消息主题；

接收模块23，用于上述消息消费者从上述消息管理者接收上述目标消息主题关联的消息，解析并消费上述消息；

请参见图7，在一种可能的实施方式中，上述解析消费模块24用于：

从上述接收模块接收的消息的头部信息中获取消息模式标识，并通过上述发送模块向消息管理者发送消息模式获取请求，上述消息模式获取请求中携带上述消息模式标识；

从上述消息管理者获取上述消息模式标识对应的消息模式，并基于上述消息模式对上述消息进行解析后消费。

在一种可能的实施方式中，上述解析消费模块24还用于：

从上述接收模块接收的消息的头部信息中获取消息模式标识，并通过上述发送模块向消息管理者发送消息模式获取请求，上述消息模式获取请求中携带上述消息模式标识；

从上述消息管理者获取上述消息模式标识对应的消息模式，并基于上述消息模式对上述消息进行解析后消费。

其中，获取模块21、发送模块22、接收模块23、解析消费模块24的具体功能实现方式可以参见上述图5对应实施例中的步骤S301-S304，这里不再进行赘述。

请一并参见图8，是本发明实施例提供的另一种基于消息队列的数据传输装置的结构示意图。如图8所示，基于消息队列的数据传输装置3可以是对应于消息生产者的用户终端，该数据传输装置3包括：创建模块31、消息生成模块32、发送模块33。

创建模块31，用于创建数据源；

消息生成模块32，根据上述创建模块创建的上述数据源生成消息和上述消息关联的数据源信息；

发送模块33，用于向消息管理者发送消息数据，上述消息数据中携带上述消息生成模块生成的上述消息和上述数据源信息。

请参见图8，在一种可行的实施方式中，上述发送模块33还用于：

向消息管理者和/或消息消费者发送上述数据源的数据源信息；

其中，上述数据源信息包括数据源名称和数据源所属项目的项目名称。

进一步地，请参见图9，图9是本申请提供的计算机设备的一结构示意图。如图9所示，该计算机设备1000可以适用于消息管理者的用户终端，该计算机设备1000可以包括：至少一个处理器1001，例如CPU，至少一个网络接口1004，收发器1003，存储器1005，至少一个通信总线1002。其中，通信总线1002用于实现这些组件之间的连接通信。其中，网络接口1004可选地可以包括标准的有线接口、无线接口(如WI-FI接口)。存储器1005可以是高速随机存储记忆体(random access memory，RAM)存储器，也可以是非不稳定的存储器(non-volatile memory)，例如至少一个磁盘存储器。存储器1005可选地还可以是至少一个位于远离前述处理器1001的存储装置。如图9所示，作为一种计算机存储介质的存储器1005中可以包括操作系统、网络通信模块、用户接口模块以及设备控制应用程序。

在图9所示的计算机设备1000中，网络接口1004主要用于用户终端进行网络通信，处理器1001和收发器1003可以用于调用存储器1005中存储的设备控制应用程序，以执行如下操作：

上述收发器1003，用于接收消息生产者发送的消息数据，消息数据中携带上述消息生产者生成的消息和消息关联的数据源信息；

上述处理器1001，用于确定数据源信息关联的消息队列，并根据数据源信息生成目标消息主题；

上述处理器1001，还用于在上述消息队列中确定出目标消息主题关联的目标存储区域，并将目标消息主题和消息存储至目标存储区域；

当上述收发器1003接收到消息消费者发送的消息订阅请求且消息订阅请求中携带数据源信息时，上述处理器1001，还用于根据数据源信息从消息队列的目标存储区域中确定出目标消息主题，从目标存储区域中拉取目标消息主题对应的消息并向消息消费者推送消息。

在一些可行的实施方式中，上述消息队列中包括上述目标存储区域在内的至少一个存储区域，一个存储区域关联一个消息主题并用于存储上述消息主题的消息，同一个消息主题的消息关联的数据源信息相同。

在一些可行的实施方式中，上述数据源信息中包括上述消息所属数据源的数据源名称和上述数据源所属项目的项目名称，上述数据源所属项目中包括上述数据源在内的至少一个数据源；上述处理器1001，还用于根据上述项目名称和上述数据源名称生成目标消息主题，其中，一个数据源名称对应一个消息主题。

在一些可行的实施方式中，上述处理器1001，还用于将上述消息模式注册到消息注册服务中心，并通过上述消息注册服务中心生成上述消息的消息模式对应的消息模式标识；

上述收发器1003，还用于当接收到上述消息消费者发送的消息模式获取请求且上述消息模式获取请求中携带上述消息模式标识后，从上述消息注册服务中心获取上述消息模式标识对应的上述消息模式向上述消息消费者发送上述消息模式。

应当理解，本申请中所描述的计算机设备1000可执行前文图4所对应实施例中对该基于消息队列的数据传输方法的描述，也可执行前文图6所对应实施例中对该基于消息队列的数据传输装置1的描述，在此不再赘述。另外，对采用相同方法的有益效果描述，也不再进行赘述。

此外，这里需要指出的是：本申请还提供了一种计算机可读存储介质，且该计算机可读存储介质中存储有前文提及的基于消息队列的数据传输装置1所执行的计算机程序，且该计算机程序包括程序指令，当该处理器执行该程序指令时，能够执行前文图4所对应实施例中对该基于消息队列的数据传输方法的描述，因此，这里将不再进行赘述。

进一步的，请参见图10是本发明实施例提供的另一种计算机设备的结构示意图。如图10所示，该计算机设备2000可以适用于消息消费者的用户终端，该计算机设备2000可以包括：至少一个处理器2001，例如CPU，至少一个网络接口2004，收发器2003，存储器2005，至少一个通信总线2002。其中，通信总线2002用于实现这些组件之间的连接通信。其中，网络接口2004可选地可以包括标准的有线接口、无线接口(如WI-FI接口)。存储器2005可以是高速随机存储记忆体(random access memory，RAM)存储器，也可以是非不稳定的存储器(non-volatile memory)，例如至少一个磁盘存储器。存储器2005可选地还可以是至少一个位于远离前述处理器2001的存储装置。如图10所示，作为一种计算机存储介质的存储器2005中可以包括操作系统、网络通信模块、用户接口模块以及设备控制应用程序。

在图10所示的计算机设备2000中，网络接口2004主要用于用户终端进行网络通信；而处理器2001和收发器2003可以用于调用存储器2005中存储的设备控制应用程序，以执行如下操作：

上述收发器2003，用于获取消息生产者创建的数据源的数据源信息；

上述收发器2003，还用于当请求订阅上述消息生产者生成的消息时，向消息管理者发送消息订阅请求，上述消息订阅请求中携带数据源信息，上述数据源信息用于确定上述消息消费者请求订阅的目标消息主题；

当上述收发器2003从上述消息管理者接收上述目标消息主题关联的消息后，上述处理器2001，用于解析并消费上述消息。

在一种可行的实施方式中，上述收发器2003，还用于从上述消息的头部信息中获取消息模式标识信息；

上述处理器2001，还用于根据上述消息模式标识信息确定上述消息的消息模式，并基于上述消息的消息模式对上述消息进行解析后消费。

在一种可行的实施方式中，上述收发器2003，还用于从上述消息的头部信息中获取消息模式标识，并向消息管理者发送消息模式获取请求，上述消息模式获取请求中携带上述消息模式标识；从上述消息管理者获取上述消息模式标识对应的消息模式；

上述处理器2001，还用于基于上述消息模式对上述消息进行解析后消费。

应当理解，本申请中所描述的计算机设备2000可执行前文图5所对应实施例中对该基于消息队列的数据传输方法的描述，也可执行前文图7所对应实施例中对该基于消息队列的数据传输装置2的描述，在此不再赘述。另外，对采用相同方法的有益效果描述，也不再进行赘述。

此外，这里需要指出的是：本申请还提供了一种计算机可读存储介质，且该计算机可读存储介质中存储有前文提及的基于消息队列的数据传输装置2所执行的计算机程序，且该计算机程序包括程序指令，当该处理器执行该程序指令时，能够执行前文图5所对应实施例中对该基于消息队列的数据传输方法的描述，因此，这里将不再进行赘述。

进一步的，请参见图11是本发明实施例提供的另一种计算机设备的结构示意图。如图11所示，该计算机设备3000可以适用于消息生产者的用户终端，该计算机设备3000可以包括：至少一个处理器3001，例如CPU，至少一个网络接口3004，收发器3003，存储器3005，至少一个通信总线3002。其中，通信总线3002用于实现这些组件之间的连接通信。其中，网络接口3004可选地可以包括标准的有线接口、无线接口(如WI-FI接口)。存储器3005可以是高速随机存储记忆体(random access memory，RAM)存储器，也可以是非不稳定的存储器(non-volatile memory)，例如至少一个磁盘存储器。存储器3005可选地还可以是至少一个位于远离前述处理器3001的存储装置。如图11所示，作为一种计算机存储介质的存储器3005中可以包括操作系统、网络通信模块、用户接口模块以及设备控制应用程序。

在图11所示的计算机设备3000中，网络接口3004主要用于用户终端进行网络通信；而处理器3001和收发器3003可以用于调用存储器3005中存储的设备控制应用程序，以执行如下操作：

上述处理器3001，用于创建数据源，并根据上述数据源生成消息和上述消息关联的数据源信息；

上述收发器3003，用于向消息管理者发送消息数据，上述消息数据中携带所述消息和所述数据源信息。

在一种可行的实施方式中，上述收发器3003，还用于在处理器3001创建数据源后，向消息管理者和/或消息消费者发送上述数据源的数据源信息，其中，上述数据源信息包括数据源名称和数据源所属项目的项目名称。

应当理解，本申请中所描述的计算机设备3000可执行前文图3所对应实施例中对该基于消息队列的数据传输方法的描述，也可执行前文图8所对应实施例中对该基于消息队列的数据传输装置3的描述，在此不再赘述。另外，对采用相同方法的有益效果描述，也不再进行赘述。

此外，这里需要指出的是：本申请还提供了一种计算机可读存储介质，且该计算机可读存储介质中存储有前文提及的基于消息队列的数据传输装置3所执行的计算机程序，且该计算机程序包括程序指令，当该处理器执行该程序指令时，能够执行前文图3所对应实施例中对该基于消息队列的数据传输方法的描述，因此，这里将不再进行赘述。

另外，对采用相同方法的有益效果描述，也不再进行赘述。对于本申请所涉及的计算机可读存储介质实施例中未披露的技术细节，请参照本申请方法实施例的描述。作为示例，程序指令可被部署为在一个计算设备上执行，或者在位于一个地点的多个计算设备上执行，又或者，在分布在多个地点且通过通信网络互连的多个计算设备上执行，分布在多个地点且通过通信网络互连的多个计算设备可以组成区块链系统。

本领域普通技术人员可以理解实现上述实施例方法中的全部或部分流程，是可以通过计算机程序来指令相关的硬件来完成，上述的程序可存储于一计算机可读取存储介质中，该程序在执行时，可包括如上述各方法的实施例的流程。其中，上述计算机可读存储介质可以是前述任一实施例提供的基于消息队列的数据传输装置或者上述设备的内部存储单元，例如电子设备的硬盘或内存。该计算机可读存储介质也可以是该电子设备的外部存储设备，例如该电子设备上配备的插接式硬盘，智能存储卡(smart media card,SMC)，安全数字(secure digital,SD)卡，闪存卡(flash card)等。上述计算机可读存储介质还可以包括磁碟、光盘、只读存储记忆体(read-only memory，ROM)或随机存储记忆体等。进一步地，该计算机可读存储介质还可以既包括该电子设备的内部存储单元也包括外部存储设备。该计算机可读存储介质用于存储该计算机程序以及该电子设备所需的其它程序和数据。该计算机可读存储介质还可以用于暂时地存储已经输出或者将要输出的数据。

本发明的权利要求书和说明书及附图中的术语“包括”和“具有”以及它们任何变形，意图在于覆盖不排他的包含。例如包含了一系列步骤或单元的过程、方法、系统、产品或设备没有限定于已列出的步骤或单元，而是可选地还包括没有列出的步骤或单元，或可选地还包括对于这些过程、方法、产品或设备固有的其它步骤或单元。在本文中提及“实施例”意味着，结合实施例描述的特定特征、结构或特性可以包含在本发明的至少一个实施例中。在说明书中的各个位置展示该短语并不一定均是指相同的实施例，也不是与其它实施例互斥的独立的或备选的实施例。本领域技术人员显式地和隐式地理解的是，本文所描述的实施例可以与其它实施例相结合。在本发明说明书和所附权利要求书中使用的术语“和/或”是指相关联列出的项中的一个或多个的任何组合以及所有可能组合，并且包括这些组合。

本领域普通技术人员可以意识到，结合本文中所公开的实施例描述的各示例的单元及算法步骤，能够以电子硬件、计算机软件或者二者的结合来实现，为了清楚地说明硬件和软件的可互换性，在上述说明中已经按照功能一般性地描述了各示例的组成及步骤。专业技术人员可以对每个特定的应用来使用不同方法来实现所描述的功能，但是这种实现不应认为超出本发明的范围。

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

Claims (10)

1.一种基于消息队列的数据传输方法，其特征在于，所述方法包括：

消息管理者接收消息生产者发送的消息数据，所述消息数据中携带所述消息生产者生成的消息和所述消息关联的数据源信息；

所述消息管理者确定所述数据源信息关联的消息队列，并根据所述数据源信息生成目标消息主题；

所述消息管理者在所述消息队列中确定出所述目标消息主题关联的目标存储区域，并将所述目标消息主题和所述消息存储至所述目标存储区域；

当接收到消息消费者发送的消息订阅请求且所述消息订阅请求中携带所述数据源信息时，所述消息管理者根据所述数据源信息从所述消息队列的所述目标存储区域中确定出所述目标消息主题，从所述目标存储区域中拉取所述目标消息主题对应的所述消息并向所述消息消费者推送所述消息。

2.根据权利要求1所述的方法，其特征在于，所述消息队列中包括所述目标存储区域在内的至少一个存储区域，一个存储区域关联一个消息主题并用于存储所述消息主题的消息，同一个消息主题的消息关联的数据源信息相同。

3.根据权利要求1或2所述的方法，其特征在于，所述数据源信息中包括所述消息所属数据源的数据源名称和所述数据源所属项目的项目名称，所述数据源所属项目中包括所述数据源在内的至少一个数据源；

所述根据所述数据源信息生成目标消息主题包括：

根据所述项目名称和所述数据源名称生成目标消息主题，其中，一个数据源名称对应一个消息主题。

4.根据权利要求1-3任一项所述的方法，其特征在于，所述消息管理者接收消息生产者发送的消息数据之后，所述方法还包括：

所述消息管理者将所述消息的消息模式注册到消息注册服务中心，通过所述消息注册服务中心生成所述消息的消息模式对应的消息模式标识；

当接收到所述消息消费者发送的消息模式获取请求且所述消息模式获取请求中携带所述消息模式标识时，从所述消息注册服务中心获取所述消息模式标识对应的所述消息模式并向所述消息消费者发送所述消息模式。

5.一种基于消息队列的数据传输方法，其特征在于，所述方法包括：

消息消费者获取消息生产者创建的数据源的数据源信息；

当所述消息消费者请求订阅所述消息生产者生成的消息时，向消息管理者发送消息订阅请求，所述消息订阅请求中携带所述数据源信息，所述数据源信息用于确定所述消息消费者请求订阅的目标消息主题；

所述消息消费者从所述消息管理者接收所述目标消息主题关联的消息，解析并消费所述消息。

6.权利要求5所述的方法，其特征在于，所述解析并消费所述消息，包括：

所述消息消费者从所述消息的头部信息中获取消息模式标识信息，并根据所述消息模式标识信息确定所述消息的消息模式；

基于所述消息的消息模式对所述消息进行解析后消费。

7.根据权利要求5所述的方法，其特征在于，所述解析并消费所述消息，包括：

所述消息消费者从所述消息的头部信息中获取消息模式标识，并向消息管理者发送消息模式获取请求，所述消息模式获取请求中携带所述消息模式标识；

从所述消息管理者获取所述消息模式标识对应的消息模式，并基于所述消息模式对所述消息进行解析后消费。

8.一种基于消息队列的数据传输方法，其特征在于，所述方法包括：

消息生产者创建数据源，并根据所述数据源生成消息和所述消息关联的数据源信息；

所述消息生产者向消息管理者发送消息数据，所述消息数据中携带所述消息和所述数据源信息。

9.一种计算机设备，其特征在于，包括：收发器、处理器、存储器以及网络接口；

所述处理器与所述收发器、所述存储器、网络接口相连，所述网络接口用于提供数据通信功能，所述存储器用于存储程序代码，所述收发器和所述处理器用于调用所述程序代码，执行权利要求1-8任一项所述的方法。

10.一种计算机可读存储介质，其特征在于，所述计算机可读存储介质存储有计算机程序，所述计算机程序包括程序指令，所述程序指令被处理器执行时，执行权利要求1-8任一项所述的方法。

Images