CN112689248B 一种消息处理方法及系统 （深圳前海微众银行股份有限公司）

A,2020.02.04A,2020.11.17A,2018.10.09A,2020.11.24本发明实施例提供了一种消息处理方法及可以避免对消息服务端在消息处理过程中所涉2第一消息端通过第一订阅机制，将请求消息发送至消息服务所述消息服务端根据所述第一订阅主题，将所述请求消息存储至所所述第二消息端通过所述第一订阅机制，从所述消息服务端获取所述请所述消息服务端根据所述响应消息中的第二订阅主题，将所述二订阅主题的第二存储区；所述第二存储区为第二订阅主题下的响应队列信息指示的队所述第一消息端在所述消息服务端创建请求消息的第一订阅主题及请求消息的响应所述第一消息端通过服务注册中心确定所述第二订阅主题所述第一消息端从所述第二订阅主题下的队列信息中确定所述响应消息的响应队列所述第一消息端根据自身的应用实例的个数和所述第二订阅主题下的队列信息中的所述第二消息端生成所述第一订阅主题对应的消息所述第二消息端通过所述第一订阅机制将所述消息拉取请求发送给所述消息服务端；所述消息拉取请求用于指示所述消息服务端基于所述第一订阅主题从所述第一存储区中所述消息服务端通过回放线程确定出所述第一存储区中各待解析消所述消息服务端基于所述各待解析消息的位置，对所述各待解析消息进行解析处理，3端发送的消息拉取请求对应的消息偏移量从所述第一存储区中查所述第一处理单元，用于通过第二订阅机制，监听所述第二订阅所述第二处理单元，用于通过所述第一订阅机制，从所述消息服务端获所述第三处理单元，用于根据所述第一订阅主题，将所述请求所述第一处理单元还用于：在所述消息服务端创建请求消息的第一订阅从所述第二订阅主题下的队列信息中确定所述响应消息的45[0015]所述第一消息端在所述消息服务端创建请求消息的第一订阅主题及请求消息的[0017]所述第一消息端从所述第二订阅主题下的队列信息中确定所述响应消息的响应一消息端可以及时准确地接收到请求消息对应的响应消息，并可以避免响应消息的丢失，[0020]所述第一消息端根据自身的应用实例的个数和所述第二订阅主题下的队列信息[0024]所述第二消息端通过所述第一订阅机制将所述消息拉取请求发送给所述消息服6述各待解析消息为所述第一存储区中待解析消息端发送的消息拉取请求对应的消息偏移量从所述第一存储区中查询出对应的请求消于消息拉取请求对应的消息偏移量快速准确[0044]在所述消息服务端创建请求消息的第一订阅主题及请求消息的响应消息的第二7取请求用于指示所述消息服务端基于所述第一订阅主题从所述第一存储区中确定出所述息拉取请求对应的消息偏移量从所述第一存储区中查询出8[0069](4)NameServer：是一个轻量级的服务注册中心。每个NameServer节点中有所有Broker中topic的路由信息，RocketMQ架构中broker与所有的NameServer维持长连接，Producer和Consumer与任一个NameServe[0072](7)Message：是消息的载体，一个Message必须指定topiProducer发送消息和Consumer接收消息必须预[0073](8)MessageQueue：消息队列，topic的细分，多个Queue从逻辑上组成了一个[0077](12)RemotingCommand：是RocketMQ进行远程交互的消息封装，它包含code、9码，topic管理有topic管理对应的交易码，不同的交易对应了不同的交易码。服务端(指Broker)或者客户端(指Producer或者Consumer)在启动时，会同时初始化并启动remoting[0078](13)Rocket-Remoting：remoting模块是服务端或者客户端用来进行消息发送和息消费者可以根据订阅需求从消息服务端的topic中拉取与订阅需求对应的标注后的消应答消息丢失的概率增加，并无法对应答消息进行重放。而且还需要进行多次修改代码，的特殊topic。(2)修改Broker端代码，扩展发送消息处理器并注册到remoting模块中。remoting模块接收到Producer模块的请求消息解析成RemotingCommand，获取到交易码取消息头的REPLY_TO标识，并把这个唯一标识和Producer对应的SocketChannel保存到Broker端代码，新增回包消息处理器并注册到remoting模块中。remoting模块接收到过这个唯一标识从内存映射表producerChannelTable中找到原请求消息的SocketChannel[0085]为了便于理解本发明实施例，首先以图1中示出的系统架构为例说明适用于本发SystemA-REPLY-TOPIC，并在启动时订阅用于业务系统的Topic(比如TopicA、TopciB、TopicC等)和用于接收响应消息的replyTopic。之后生产出与topic对应的消息(比如与TopicA对应的消息、与TopciB对应的消息、与TopicC对应的消息等)，并将该消息标识上务端120基于消息拉取请求将TopicA存储的消息发送给第二消息端130，第二消息端130基消息以及TopicA，生成响应消息，将响应消息发送给消息服务端120的SystemA-REPLY-息服务端120发送响应消息拉取请求，以使消息服务端120基于响应消息拉取请求将SystemA-REPLY-TOPIC响应消息队列存储的TopicA对应的响应消息发送给第一消息端110。[0088]基于上述描述，图2示例性的示出了本发明实施例提供的一种消息处理方法的流身的应用实例的个数和第二订阅主题下的队列信息中的队列数量，按照平均分配的方式，的队列信息，并可以避免某一应用实例因负载压力过大而出现故障或数据处理缓慢的情它与所有拥有Topic的broker进行连接，在发送请求消息时采用轮询机制发送到broker息服务端根据第二消息端发送的消息拉取请求对应的消息偏移量从第一存储区中查询出个请求消息的偏移量，将该多个请求消息的偏移量存储在与第一订阅主题对应的已有Offset偏移量(即上一次解析完成时的消息位置)，新进入CommitLog的消息会追加在CommitLog的末尾，回放线程若确定CommitLog文件的长度大于已有Offset偏移量，则从Offset处开始解析每一条消息并放到消息对应topic的Consume而可以避免对消息服务端在消息处理过程中所涉及的代[0103]Producer端在发送消息到Broker端时，需要在消息头附带上当前系统监听的[0105]每个系统(通常指Producer生产者和Consumer消费者的组合)接入时，都需要在Broker端新建一个用于接收响应消息的replyTopic。比如SystemA(Producer生产者和[0107]一个应用于生产环境的系统，通常是Producer生产者和Consumer消费者的组个topic可以由同一个消费者组ConsumerGroup(也可以指消费者集群)下所有消费者平均的队列消息通常由同一个消费者组ConsumerGroup下所有消费者平端从该topic所属的ConsumeQueue中查找出消息，并返回对应的消息给Consumer端，[0110]为了确保应答消息能够被发出请求消息的Producer端实例接收到，则需要Producer端在发送请求消息时，在消息头上附带该Producer端订阅的replyTopic对应的变的。可以根据这个原则在发送请求消息时附带上这个实例订阅的replyTopic的MessageQueue信息，下游收到请求回包应答消息到当前系统监听的MessageQueue上面来，这样就可以确保应答消息能够被当前系统正常接收到。用平均消息分配策略算法进行初始化，然后通过该consumer实例从NameServer端获取衡算法从当前实例监听的消息队列MessageQueue列表中随机选择一个作为回报消息的这样以便接收到此消息的下游子系统能够解析消息头并回包应答消息到Broker端应的业务处理。并在处理完业务之后，获取请求消息头的REPLY_TOPIC、REPLY_BROKER、[0117]基于相同的技术构思，图3示例性的示出了本发明实施例提供的一种消息处理系[0129]在所述消息服务端创建请求消息的第一订阅主题及请求消息的响应消息的第二取请求用于指示所述消息服务端基于所述第一订阅主题从所述第一存储区中确定出所述息拉取请求对应的消息偏移量从所述第一存储区中查询出器401与存储器402之间的具体连接介质，图4中处理器401和存储器402之间通过总线连接[0146]存储器402作为一种非易失性计算机可读存储介质，可用于存储非易失性软件程例如可以包括闪存、硬盘、多媒体卡、卡型存储器、随机访问存储器(RandomAccess电可擦除可编程只读存储器(ElectricallyErasableProgrammableRead-OnlyMemory，例中的存储器402还可以是电路或者其它任意能够实现存储功能的装置，用于存储程序指程图