一种消息处理方法、装置、设备及存储介质与流程

1.本发明涉及大数据领域，尤其涉及一种消息处理方法、装置、设备及存储介质。


背景技术：

2.在消息系统的一些业务中，存在一些场景，需要对生产者发布的消息内容做一些加工处理，然后再由消费者来接收并消费。针对该场景，目前在使用消息队列的系统中一般使用以下几种解决方案，但均存在一些缺陷，如，消费者从服务节点读取原始消息后，由消费者所在的客户端对原始消息进行内容加工，采用该方案，客户端接收的原始消息一方面可能包含消费者不需要的内容，会造成网络带宽的浪费，另一方面可能包含不希望发给客户端的敏感信息，存在安全隐患。或者在接受到生产者发送的原始消息后，通过新建主题来加工并存储后的消息，采用该方案需要消耗额外的存储空间，且由于新建主题对应的加工节点需要从服务节点消费原始消息，加工完成后再反馈给服务节点，还会造成带宽浪费，会降低消息系统的吞吐量。


技术实现要素：

3.为解决上述技术问题，本技术实施例期望提供一种消息处理方法、装置、设备及存储介质。
4.本技术的技术方案是这样实现的：
5.第一方面，提供了一种消息处理方法，所述方法包括：
6.为目标主题在服务节点配置主数据管道和加工数据管道；其中，所述目标主题为需要进行内容加工的主题；
7.将所述目标主题对应的生产者发送的原始消息写入所述主数据管道，并对所述原始消息进行存储；
8.异步触发针对所述原始消息的加工处理，得到加工消息；
9.将所述加工消息写入所述加工数据管道，并对所述加工消息进行存储。
10.上述方案中，所述方法还包括：确定第一消费者读取的消息为所述目标主题的原始消息时，指示所述第一消费者订阅所述主数据通道；确定第二消费者读取的消息为所述目标主题的加工消息时，指示所述第二消费者订阅所述加工数据管道。
11.上述方案中，所述方法还包括：存储所述原始消息的标识信息与所述加工消息的标识信息之间的对应关系；确定第一加工消息丢失时，基于所述对应关系和所述第一加工消息的标识信息，确定所述第一加工消息对应的第一原始消息的标识信息；基于所述第一原始消息的标识信息，从所述主数据管道中读取所述第一原始消息；对读取到的所述第一原始消息进行加工处理，得到所述第一加工消息；将所述第一加工消息写入所述加工数据管道，并对所述第一加工消息进行存储。
12.上述方案中，所述对所述原始消息进行存储，包括：对所述原始消息以多副本形式进行存储。
13.上述方案中，所述对所述加工消息进行存储，包括：对所述加工消息以单副本形式进行存储。
14.上述方案中，所述方法还包括：所述原始消息存储完成后，向所述生产者发送接收成功消息。
15.第二方面，提供了一种消息处理装置，所述装置包括：
16.配置模块，用于为目标主题在服务节点配置主数据管道和加工数据管道；其中，所述目标主题为需要进行内容加工的主题；
17.处理模块，用于将所述目标主题对应的生产者发送的原始消息写入所述主数据管道，并对所述原始消息进行存储；
18.所述处理模块，还用于异步触发针对所述原始消息的加工处理，得到加工消息；将所述加工消息写入所述加工数据管道，并对所述加工消息进行存储。
19.第三方面，提供了一种电子设备，所述电子设备包括：处理器和配置为存储能够在处理器上运行的计算机程序的存储器，其中，所述处理器配置为运行所述计算机程序时，执行前述方法的步骤。
20.第四方面，提供了一种计算机存储介质，其上存储有计算机程序，其中，该计算机程序被处理器执行时实现前述方法的步骤。
21.第五方面，提供了一种消息系统，所述消息系统包括：生产者、服务节点和消费者；所述服务节点包括：主题管理器、数据加工管理器和存储器；
22.所述生产者，用于发送原始消息至所述服务节点；
23.所述主题管理器，用于确定目标主题为需要进行内容加工的主题时，为所述目标主题配置主数据管道和加工数据管道；将所述生产者发送的所述原始消息写入所述主数据管道；
24.所述数据加工管理器，用于异步触发数据加工器对所述原始消息进行加工处理，得到加工消息；将所述加工消息写入所述加工数据管道；
25.所述存储器，用于对所述主数据管道和所述加工数据管道中的消息进行存储；
26.所述消费者，用于从所述主数据管道读取所述原始消息，或者从所述加工数据管道中读取所述加工消息。
27.本技术公开一种消息处理方法、装置、设备及存储介质，通过为主题配置主数据管道和加工数据管道，以异步的方式对原始消息进行加工及存储，使得该主题的消费者可以直接从服务节点的主数据管道、加工数据管道中读取到原始消息和加工消息。本技术技术方案，相比于由消费者所在客户端进行消息加工的方案，解决了需要将原始消息全量传给客户端带来的额外网络资源消耗和数据安全性问题；相比于新建主题来加工并存储加工消息的方案，解决了原始消息和加工消息需要在不同服务节点和加工节点之间传输所带来的网络损耗问题，提高了消息系统的吞吐量。
附图说明
28.图1为本技术实施例中消息处理方法的第一流程示意图；
29.图2为本技术实施例中消息处理装置的组成结构示意图；
30.图3为本技术实施例中电子设备的组成结构示意图；
31.图4为本技术实施例中消息系统的结构示意图；
32.图5为本技术实施例中生产、加工及消费消息的具体流程示意图；
33.图6为本技术实施例中恢复加工消息的流程示意图。
具体实施方式
34.为了能够更加详尽地了解本技术实施例的特点与技术内容，下面结合附图对本技术实施例的实现进行详细阐述，所附附图仅供参考说明之用，并非用来限定本技术实施例。
35.本技术实施例提出一种消息处理方法。图1为本技术实施例中消息处理方法的第一流程示意图，如图1所示，该消息处理方法具体可以包括：
36.步骤101：为目标主题在服务节点配置主数据管道和加工数据管道；其中，所述目标主题为需要进行内容加工的主题；
37.这里，服务节点为目标主题对应的服务器节点，可以接收生产者传输的消息数据，并向消费者提供消息数据。示例性的，实际应用中，服务节点可以为broker节点。
38.实际应用中，每个主题在服务节点可以配置多个数据管道。数据管道是用来存储消息内容的有序队列。主数据管道用于存储生产者发送的原始消息(也可称原始数据)，加工数据管道用于存储对原始消息进行加工后的加工消息(也可称加工数据)。示例性的，采用不同的加工方式对原始消息进行加工，得到不同的加工消息。一个主数据管道可以对应一个或多个加工数据管道，用于存储通过不同加工方式加工得到的加工消息。
39.示例性的，一个主题中，包含一个主数据管道，为可写(存储生产者发送的原始数据)并且可读(消费者可指定该数据管道并从中消费数据)，包含一个或多个加工数据管道，只用于存储加工后的消息内容，为可读。
40.步骤102：将所述目标主题对应的生产者发送的原始消息写入所述主数据管道，并对所述原始消息进行存储；
41.这里，生产者为用于生产消息的模块，实际应用中，生产者可以为发布消息的应用；消费者为用于消费消息的模块，实际应用中，消费者可以为订阅消息的应用。
42.这里，原始消息为目标主题对应的生产者直接发送的消息。通过将原始消息写入主数据管道，可以使得该主题的消费者可以直接从主数据管道中读取到原始消息。
43.示例性的，在一些实施例中，所述对所述原始消息进行存储，包括：对所述原始消息以多副本形式进行存储。
44.通过以多副本形式对原始消息进行存储，可以减小原始消息的丢失概率。
45.示例性的，在一些实施例中，所述方法还包括：所述原始消息存储完成后，向所述生产者发送接收成功消息。
46.由于加工处理是异步进行的，目标主题对应的服务节点(如broker节点)接收消息时只需要存储原始数据，不需要进行加工处理就可以给生产者反馈接收成功的通知，让生产者发送下一批消息，从而提高了生产消息时的消息吞吐量。
47.步骤103：异步触发针对所述原始消息的加工处理，得到加工消息；
48.这里，对所述原始消息的加工处理，可以是对消息内容的编码格式做一些转换，让消息可以由不同协议的消费客户端来接收并消费。比如通过开源流处理平台kafka客户端发送kafka协议编码的消息体，在进行了消息编码转换到云原生分布式消息流平台pulsar
协议后由pulsar客户端接收并消费。也可以是对消息内容进行加工处理，如对原始消息中的某些字段做脱敏处理，去掉一些敏感信息(如用户名，密码等)。
49.实际应用中，对所述原始消息的加工处理，可以是通过数据加工器对原始消息进行加工处理实现。示例性的，异步触发针对原始消息的加工处理，具体包括：控制数据加工器以异步的方式从主数据管道中读取原始消息；对读取到的原始消息进行加工处理，得到加工消息。
50.需要说明的是，通过不同数据加工方式加工得到的加工消息，存储于不同的加工数据管道中。示例性的，以第一加工方式对原始消息进行加工，得到第一加工消息；将所述加工消息写入第一加工数据管道进行存储；以第二加工方式对原始消息进行加工，得到第二加工消息；将所述加工消息写入第二加工数据管道进行存储。
51.示例性的，每个主题可以添加多个数据加工器。每个数据加工器包含一个用于存储加工后数据的加工数据管道名称，以及一个数据加工函数。数据加工函数为用于加工消息的函数。示例性的，一个数据加工函数对应一个加工数据管道。
52.示例性的，在实际应用中，加工处理可以包括对编码格式的加工处理，通过在服务节点上对原始消息的编码格式进行加工，使得消息系统可以支持不同协议的原生客户端程序。
53.步骤104：将所述加工消息写入所述加工数据管道，并对所述加工消息进行存储。
54.通过将加工消息写入加工数据管道，可以使得该主题的消费者可以直接从加工数据管道中读取到加工消息。虽然增加了用于存储加工后数据的存储，但是消费者可以选择消费原始数据还是加工后的加工数据，并且每份数据只会加工一次，当有多个消费者同时消费同一份加工后数据时不会对原始数据多次加工。
55.示例性的，在一些实施例中，所述对所述加工消息进行存储，包括：对所述加工消息以单副本形式进行存储。
56.通过将加工消息以单副本形式进行存储，可以减少存储加工消息所需要的存储空间。
57.实际应用中，目标主题的消费者消费消息时，可以从目标主题的主数据管道消费原始消息，也可以从目标主题的加工数据管道消费加工后的加工消息。
58.示例性的，在一些实施例中，所述方法还包括：确定第一消费者读取的消息为所述目标主题的原始消息时，指示所述第一消费者订阅所述主数据通道；确定第二消费者读取的消息为所述目标主题的加工消息时，指示所述第二消费者订阅所述加工数据管道。
59.这里，指示所述第一消费者订阅所述主数据通道，以使得所述第一消费者从所述主数据通道中读取所述原始消息；指示所述第二消费者订阅所述加工数据管道，以使得所述第二消费者从所述加工数据管道中读取所述加工消息。
60.示例性的，在一些实施例中，所述方法还包括：存储所述原始消息的标识信息与所述加工消息的标识信息之间的对应关系；确定第一加工消息丢失时，基于所述对应关系和所述第一加工消息的标识信息，确定所述第一加工消息对应的第一原始消息的标识信息；基于所述第一原始消息的标识信息，从所述主数据管道中读取所述第一原始消息；对读取到的所述第一原始消息进行加工处理，得到所述第一加工消息；将所述第一加工消息写入所述加工数据管道，并对所述第一加工消息进行存储。
61.示例性的，原始消息、加工消息的标识信息可以为消息id，可以设置一个可持久化存储的索引，用于存储加工消息id和原始消息id的对应关系。当加工数据管道中的数据丢失时，通过触发消息恢复机制，从主数据管道中读取对应消息的原始数据并重新进行加工和存储，可有效应对加工消息丢失的情况。
62.这里，步骤101至步骤104的执行主体可以为实现本技术实施例中消息处理方法的电子设备的处理器。
63.本技术实施例的技术方案，通过为目标主题在服务节点配置主数据管道和加工数据管道，以异步的方式对原始消息进行加工及存储，使得该主题的消费者可以直接从服务节点的主数据管道、加工数据管道中读取到原始消息和加工消息。本技术技术方案，相比于由消费者所在客户端进行消息加工的方案，解决了需要将原始消息全量传给客户端带来的额外网络资源消耗和数据安全性问题；通过在服务节点上对消息的编码格式进行加工，让消息系统可以支持不同协议的原生客户端程序；相比于通过流处理的方式新建主题来加工并存储加工消息的方案，解决了原始消息和加工消息需要在不同服务节点和加工节点之间传输所带来的网络损耗问题，提高了消息系统的吞吐量。
64.图2为本技术实施例中消息处理装置的组成结构示意图，展示了一种消息处理方法的实现装置，该消息处理装置20具体包括：
65.配置模块201，用于为目标主题在服务节点配置主数据管道和加工数据管道；其中，所述目标主题为需要进行内容加工的主题；
66.处理模块202，用于将所述目标主题对应的生产者发送的原始消息写入所述主数据管道，并对所述原始消息进行存储；
67.所述处理模块202，还用于异步触发针对所述原始消息的加工处理，得到加工消息；将所述加工消息写入所述加工数据管道，并对所述加工消息进行存储。
68.本技术实施例的技术方案，通过为目标主题在服务节点配置主数据管道和加工数据管道，以异步的方式对原始消息进行加工及存储，使得该主题的消费者可以直接从服务节点的主数据管道、加工数据管道中读取到原始消息和加工消息。本技术技术方案，相比于由消费者所在客户端进行消息加工的方案，解决了需要将原始消息全量传给客户端带来的额外网络资源消耗和数据安全性问题；通过在服务节点(如broker节点)上对消息的编码格式进行加工，让消息系统可以支持不同协议的原生客户端程序；相比于通过流处理的方式新建主题来加工并存储加工消息的方案，解决了原始消息和加工消息需要在不同服务节点和加工节点之间传输所带来的网络损耗问题，提高了消息系统的吞吐量。
69.在一些实施例中，所述处理模块202，还用于确定第一消费者读取的消息为所述目标主题的原始消息时，指示所述第一消费者订阅所述主数据通道；确定第二消费者读取的消息为所述目标主题的加工消息时，指示所述第二消费者订阅所述加工数据管道。
70.在一些实施例中，所述处理模块202，还用于存储所述原始消息的标识信息与所述加工消息的标识信息之间的对应关系；确定第一加工消息丢失时，基于所述对应关系和所述第一加工消息的标识信息，确定所述第一加工消息对应的第一原始消息的标识信息；基于所述第一原始消息的标识信息，从所述主数据管道中读取所述第一原始消息；对读取到的所述第一原始消息进行加工处理，得到所述第一加工消息；将所述第一加工消息写入所述加工数据管道，并对所述第一加工消息进行存储。
71.在一些实施例中，所述处理模块202，具体用于对所述原始消息以多副本形式进行存储。
72.在一些实施例中，所述处理模块202，具体用于对所述加工消息以单副本形式进行存储。
73.在一些实施例中，所述处理模块202，还用于所述原始消息存储完成后，向所述生产者发送接收成功消息。
74.基于上述消息处理装置中各单元的硬件实现，本技术实施例还提供了一种电子设备。图3为本技术实施例中电子设备的组成结构示意图。如图3所示，该设备30包括：处理器301和配置为存储能够在处理器上运行的计算机程序的存储器302；
75.其中，处理器301配置为运行计算机程序时，执行前述实施例中的方法的步骤。
76.当然，实际应用时，如图3所示，该电子设备中的各个组件通过总线系统303耦合在一起。可理解，总线系统303用于实现这些组件之间的连接通信。总线系统303除包括数据总线之外，还包括电源总线、控制总线和状态信号总线。但是为了清楚说明起见，在图3中将各种总线都标为总线系统303。
77.在实际应用中，上述处理器可以为特定用途集成电路(asic，application specific integrated circuit)、数字信号处理装置(dspd，digital signal processing device)、可编程逻辑装置(pld，programmable logic device)、现场可编程门阵列(field－programmable gate array，fpga)、控制器、微控制器、微处理器中的至少一种。可以理解地，对于不同的设备，用于实现上述处理器功能的电子器件还可以为其它，本技术实施例不作具体限定。
78.上述存储器可以是易失性存储器(volatile memory)，例如随机存取存储器(ram，random-access memory)；或者非易失性存储器(non-volatile memory)，例如只读存储器(rom，read-only memory)，快闪存储器(flash memory)，硬盘(hdd，hard disk drive)或固态硬盘(ssd，solid-state drive)；或者上述种类的存储器的组合，并向处理器提供指令和数据。
79.在示例性实施例中，本技术实施例还提供了一种计算机可读存储介质，例如包括计算机程序的存储器，计算机程序可由电子设备的处理器执行，以完成前述消息处理方法的步骤。
80.为了能更加体现本技术的目的，在本技术上实施例的基础上，进行进一步的举例说明。图4为本技术实施例中消息系统的结构示意图，该消息系统可以为分布式消息系统。其中，分布式消息系统是一种可扩展的多节点消息队列系统。如图4所示，该消息系统40包括：生产者、服务节点和消费者；所述服务节点包括：主题管理器401、数据加工管理器402、存储器403；
81.示例性的，实际应用中，服务节点可以为broker节点。
82.如图4所示，所述生产者，用于发送原始消息至所述服务节点；
83.如图4所示，所述主题管理器401，用于确定目标主题为需要进行内容加工的主题时，为所述目标主题配置主数据管道和加工数据管道；将生产者发送的原始消息写入所述主数据管道，以使得所述目标主题对应的消费者从所述主数据管道中读取所述原始消息。
84.实际应用中，每个主题topic可以添加多个数据管道。其中只有一个主数据管道为
可写(存储生产者发送的原始数据)并且可读(消费者可指定该数据管道并从中消费数据)，其他加工数据管道只用于存储加工后的消息内容，为可读。
85.需要说明的是，通过不同数据加工器加工得到的加工消息，存储于不同的加工数据管道中。示例性的，第一数据加工器以第一加工方式对原始消息进行加工，得到第一加工消息；将所述加工消息写入第一加工数据管道进行存储；第二数据加工器以第二加工方式对原始消息进行加工，得到第二加工消息；将所述加工消息写入第二加工数据管道进行存储。
86.如图4所示，所述数据加工管理器402，用于异步触发数据加工器对原始消息进行加工处理，得到加工消息；将所述加工消息写入所述加工数据管道，以使得所述目标主题对应的消费者从所述加工数据管道中读取所述加工消息。
87.如图4所示，数据加工管理器402提供一个消息加工的管理接口，管理者可通过改接口动态添加和修改对消息内容的加工方式。数据加工管理器402，负责接收和处理数据加工器的管理命令。示例性的，一条添加数据加工器的命令包含：数据加工器对应topic的名称、用于存储加工结果的加工数据管道名称、包含了数据加工函数的脚本文件(如python脚本)或字节码文件(如java class)或其他可实现动态加载函数的文件。
88.如图4所示，所述存储器403，用于对所述主数据管道和所述加工数据管道中的消息进行存储。
89.示例性的，在一些实施例中，主数据管道的数据存储为多副本，即对原始消息进行多副本存储。通过以多副本形式对原始消息进行存储，可以减小原始消息的丢失概率。
90.示例性的，在一些实施例中，加工数据管道的数据存储为单副本。即对加工消息进行单副本存储。通过将加工消息以单副本形式进行存储，可以减少存储加工消息所需要的存储空间。
91.示例性的，在一些实施例中，服务节点还用于在所述原始消息存储完成后，向所述生产者发送接收成功消息。
92.由于加工处理是异步进行的，目标主题对应的服务节点接收消息时只需要存储原始数据，不需要进行加工处理就可以给生产者反馈接收成功的通知，让生产者发送下一批消息，可以提高生产消息时的消息吞吐量。
93.示例性的，在实际应用中，一些服务节点不具备存储功能。因此，在一些实施例中，消息系统包括：生产者、服务节点、消费者和存储层；服务节点包括：主题管理器、数据加工管理器、存储管理器，存储管理器用于将所述主数据管道和所述加工数据管道中的数据存储至所述存储层或从存储层中读取数据。
94.如图4所示，所述消费者，用于从所述主数据管道读取所述原始消息，或者从所述加工数据管道中读取所述加工消息。
95.本技术的消息系统，支持在服务节点接收到生产者发送的消息后，对消息内容以异步的方式进行加工并存储，使得消费者即可以接收原始消息内容，也可以接收加工后的消息内容。由于对消息内容的加工是在接收到消息的服务节点broker上进行的，相比于由消费者所在客户端进行消息加工的方案，解决了需要将原始消息全量传给客户端带来的额外网络资源消耗和数据安全性问题；通过在服务节点broker上对消息的编码格式进行加工，让消息系统可以支持不同协议的原生客户端程序；相比于通过流处理的方式新建主题
来加工并存储加工消息的方案，不需要通过网络传输将消息发送到其他加工节点上进行处理，不影响整个消息系统的吞吐量；由于加工处理是异步进行的，目标主题对应的服务节点broker接收消息时只需要存储原始数据，不需要进行加工处理就可以给生产者反馈接收成功的通知，让生产者发送下一批消息，可以提高生产消息时的消息吞吐量。
96.示例性的，使用该消息系统40进行消息处理的过程包括：预先配置过程；生产、加工及消费消息过程。
97.具体的，预先配置过程包括：为topica配置主数据管道、加工数据管道和数据加工器a。创建消费者a订阅topica的主数据管道，消费原始消息。创建消费者b订阅topica的加工数据管道，消费加工后的加工消息。
98.示例性的，图5为本技术实施例中生产、加工及消费消息的具体流程示意图。如图5所示，该流程具体包括：
99.服务节点broker接收生产者向对应topica发送的原始消息后，将原始消息写入主数据管道，并触发存储器将原始消息以多副本的存储至磁盘，存储完成后给生产者返回发送成功消息；数据写入主数据管道后，异步触发数据加工器a，从主数据管道中读取原始消息，进行加工处理，并把数据加工后的结果(即加工消息)写入加工数据管道，并触发存储器将数据以单副本的形式存储至磁盘，以及在数据索引中存入加工消息id和原始消息id的对应关系；消费者a从topica的主数据管道中读取原始消息；消费者b从topica的加工数据管道中读取加工后的加工消息。
100.如图5所示，消费者a从topica的主数据管道中读取原始消息，具体包括：主数据管道接收消费者a发送的消费原始消息的请求；触发存储器从磁盘中读取原始消息；读取原始消息完成后，返回原始消息至消费者a。
101.如图5所示，消费者b从topica的加工数据管道中消费加工后的加工消息，具体包括：加工数据管道接收消费者b发送的消费加工消息的请求；触发存储器从磁盘中读取加工消息；读取加工消息完成后，返回加工消息至消费者b。
102.消费者需要接收加工消息时，订阅该topic中用于存储加工后数据的加工数据管道。不同于主数据管道内的数据，加工数据管道中的数据由数据加工器以异步的方式从主数据管道中读取原始消息后进行加工处理，并存储加工消息以及和原始消息编号的映射关系。且为了减少存储使用量，只存储单副本。因此，如果发生加工数据管道中存储数据丢失的情况，则需要对加工消息进行恢复。
103.在一些实施例中，使用消息系统进行消息处理的过程还包括：加工消息恢复过程。示例性的，图6为本技术实施例中恢复加工消息的流程示意图。如图6所示，读取加工消息失败(或加工消息丢失)时，确定恢复加工消息。具体的，恢复加工消息过程具体包括：
104.步骤601：根据要恢复的加工消息的id从消息id索引中获取对应的原始消息id；
105.消息id索引为一个可持久化存储的索引，用于存储加工消息id和原始消息id的对应关系，该步骤可以由存储管理器实现。
106.步骤602：数据加工器从主数据管道中读取该原始消息id对应的原始消息；
107.步骤603：由数据加工器对读取到的原始消息进行加工处理，得到加工消息；
108.步骤604：将加工消息重新写入加工数据管道进行存储。
109.实际应用中，整个加工消息恢复过程消费者不感知。
110.本技术实施例的技术方案，通过当加工数据管道中的数据丢失时，触发消息恢复机制，从主数据管道中读取对应消息的原始数据并重新进行数据加工和存储，可有效应对加工消息丢失的情况。
111.应当理解，在本技术使用的术语是仅仅出于描述特定实施例的目的，而非旨在限制本技术。在本技术和所附权利要求书中所使用的单数形式的“一种”、“所述”和“该”也旨在包括多数形式，除非上下文清楚地表示其他含义。还应当理解，本文中使用的术语“和/或”是指并包含一个或多个相关联的列出项目的任何或所有可能组合。本技术中表述“具有”、“可以具有”、“包括”和“包含”、或者“可以包括”和“可以包含”在本文中可以用于指示存在对应的特征(例如，诸如数值、功能、操作或组件等元素)，但不排除附加特征的存在。
112.应当理解，尽管在本技术可能采用术语第一、第二、第三等来描述各种信息，但这些信息不应限于这些术语。这些术语仅用来将同一类型的信息彼此区分开，不必用于描述特定的顺序或先后次序。例如，在不脱离本发明范围的情况下，第一信息也可以被称为第二信息，类似地，第二信息也可以被称为第一信息。本技术实施例所记载的技术方案之间，在不冲突的情况下，可以任意组合。在本技术所提供的几个实施例中，应该理解到，所揭露的方法、装置和设备，可以通过其它的方式实现。以上所描述的实施例仅仅是示意性的，例如，单元的划分，仅仅为一种逻辑功能划分，实际实现时可以有另外的划分方式，如：多个单元或组件可以结合，或可以集成到另一个系统，或一些特征可以忽略，或不执行。另外，所显示或讨论的各组成部分相互之间的耦合、或直接耦合、或通信连接可以是通过一些接口，设备或单元的间接耦合或通信连接，可以是电性的、机械的或其它形式的。
113.上述作为分离部件说明的单元可以是、或也可以不是物理上分开的，作为单元显示的部件可以是、或也可以不是物理单元，即可以位于一个地方，也可以分布到多个网络单元上；可以根据实际的需要选择其中的部分或全部单元来实现本实施例方案的目的。
114.另外，在本技术各实施例中的各功能单元可以全部集成在一个处理单元中，也可以是各单元分别单独作为一个单元，也可以两个或两个以上单元集成在一个单元中；上述集成的单元既可以采用硬件的形式实现，也可以采用硬件加软件功能单元的形式实现。
115.以上所述，仅为本技术的具体实施方式，但本技术的保护范围并不局限于此，任何熟悉本技术领域的技术人员在本技术揭露的技术范围内，可轻易想到变化或替换，都应涵盖在本技术的保护范围之内。

技术特征：
1.一种消息处理方法，其特征在于，所述方法包括：为目标主题在服务节点配置主数据管道和加工数据管道；其中，所述目标主题为需要进行内容加工的主题；将所述目标主题对应的生产者发送的原始消息写入所述主数据管道，并对所述原始消息进行存储；异步触发针对所述原始消息的加工处理，得到加工消息；将所述加工消息写入所述加工数据管道，并对所述加工消息进行存储。2.根据权利要求1所述的方法，其特征在于，所述方法还包括：确定第一消费者读取的消息为所述目标主题的原始消息时，指示所述第一消费者订阅所述主数据通道；确定第二消费者读取的消息为所述目标主题的加工消息时，指示所述第二消费者订阅所述加工数据管道。3.根据权利要求1所述的方法，其特征在于，所述方法还包括：存储所述原始消息的标识信息与所述加工消息的标识信息之间的对应关系；确定第一加工消息丢失时，基于所述对应关系和所述第一加工消息的标识信息，确定所述第一加工消息对应的第一原始消息的标识信息；基于所述第一原始消息的标识信息，从所述主数据管道中读取所述第一原始消息；对读取到的所述第一原始消息进行加工处理，得到所述第一加工消息；将所述第一加工消息写入所述加工数据管道，并对所述第一加工消息进行存储。4.根据权利要求1-3任一项所述的方法，其特征在于，所述对所述原始消息进行存储，包括：对所述原始消息以多副本形式进行存储。5.根据权利要求1-3任一项所述的方法，其特征在于，所述对所述加工消息进行存储，包括：对所述加工消息以单副本形式进行存储。6.根据权利要求1-3任一项所述的方法，其特征在于，所述方法还包括：所述原始消息存储完成后，向所述生产者发送接收成功消息。7.一种消息处理装置，其特征在于，所述装置包括：配置模块，用于为目标主题在服务节点配置主数据管道和加工数据管道；其中，所述目标主题为需要进行内容加工的主题；处理模块，用于将所述目标主题对应的生产者发送的原始消息写入所述主数据管道，并对所述原始消息进行存储；所述处理模块，还用于异步触发针对所述原始消息的加工处理，得到加工消息；将所述加工消息写入所述加工数据管道，并对所述加工消息进行存储。8.一种电子设备，其特征在于，所述电子设备包括：处理器和配置为存储能够在处理器上运行的计算机程序的存储器，其中，所述处理器配置为运行所述计算机程序时，执行权利要求1至6任一项所述方法的步骤。9.一种计算机可读存储介质，其上存储有计算机程序，其特征在于，该计算机程序被处理器执行时实现权利要求1至6任一项所述的方法的步骤。10.一种消息系统，其特征在于，所述消息系统包括：生产者、服务节点和消费者；所述
服务节点包括：主题管理器、数据加工管理器和存储器；所述生产者，用于发送原始消息至所述服务节点；所述主题管理器，用于确定目标主题为需要进行内容加工的主题时，为所述目标主题配置主数据管道和加工数据管道；将所述生产者发送的所述原始消息写入所述主数据管道；所述数据加工管理器，用于异步触发数据加工器对所述原始消息进行加工处理，得到加工消息；将所述加工消息写入所述加工数据管道；所述存储器，用于对所述主数据管道和所述加工数据管道中的消息进行存储；所述消费者，用于从所述主数据管道读取所述原始消息，或者从所述加工数据管道中读取所述加工消息。

技术总结
本申请公开了一种消息处理方法、装置、设备及存储介质，该方法包括：为目标主题在服务节点配置主数据管道和加工数据管道；将目标主题对应的生产者发送的原始消息写入主数据管道，并对原始消息进行存储；异步触发针对原始消息的加工处理，得到加工消息；将加工消息写入加工数据管道，并对加工消息进行存储。如此，通过为主题配置主数据管道和加工数据管道，以异步方式对原始消息进行加工及存储，使得消费者可直接从服务节点的主数据管道、加工数据管道中读取原始消息和加工消息，解决了将原始消息全量传给客户端带来的安全性问题，以及原始消息和加工消息在不同服务节点和加工节点之间传输带来的网络损耗问题，提高了消息系统的安全性和吞吐量。安全性和吞吐量。安全性和吞吐量。


技术研发人员：王嘉凌
受保护的技术使用者：中国移动通信集团有限公司
技术研发日：2022.11.14
技术公布日：2023/9/23