CN112398748A - 基于mq的智能限流方法、装置及计算机可读介质 - Google Patents

Abstract

本发明公开了一种基于MQ的智能限流方法、装置及计算机可读介质，其中方法包括：定义消息的流量级别；创建用于限制对应流量级别的消息的至少一个流量限制器和用于对各流量限制器进行限流管理的流量限制管理器；将业务产生的消息归类为对应的流量级别；流量限制管理器根据需要限制的消息的流量级别，获取可限制该流量级别的消息的流量限制器；流量限制器对达到限流条件的消息进行限流。本发明可按照不同的业务需求，将消息划分为不同的流量级别，并且可对不同流量级别的消息进行单独限流或者按照优先级进行限流，实现了对消息的有区别限流，避免了常规的限流方法对所有消息进行无区别限流可能引发系统无法正常运行等情况的发生。

Description

基于MQ的智能限流方法、装置及计算机可读介质

技术领域

本发明涉及计算机技术领域，具体涉及一种基于MQ的智能限流方法、装置及计算机可读介质。

背景技术

MQ（Message Queue），即消息队列引擎，其把传输的数据（消息）放在队列中，用队列机制来实现消息的传递——生产者产生消息并把消息放入队列，消费者到指定队列拉取消息，或者订阅相应的队列，由MQ服务端给消费者推送消息。

限流就是限制系统输入和输出的消息流量，以达到保护系统的目的。一般而言，系统的吞吐量是可以被测算的，在流量达到过载阈值时，需要采取一些限制流量的措施，比如数据延迟处理、拒绝处理或者部分拒绝处理等，以保证系统稳定运行。

令牌桶算法是目前使用较多地一种限流方法。图10示出了令牌桶算法的限流原理图，如图10所示，令牌桶算法进行限流的工作原理为：通过一个可存放固定容量令牌（token）的桶（假设桶最多可存放b个令牌），按照固定速率往桶里添加令牌，比如每秒往桶里放入10个令牌，当桶满时，新添加的令牌被丢弃或拒绝放入。当一个n字节大小的消息包到达时，将从桶中删除n个令牌，接着消息包被直接使用。如果桶中的令牌数量不足n个，则不会从桶里删除令牌，且该消息包将被限流（通常要么被丢弃，要么放入缓冲区等待）而无法被直接使用。

现有的限流方法比如令牌桶算法在限流时存在以下两个问题：

一是只能对消息进行无差别的限流处理，只要消息达到限流条件，消息将无区别的被限流；

二是被限流（比如被丢弃）的消息中如果存在关键消息，这些关键消息被丢弃后会对业务产生不利影响，系统可能无法正常运行或者某些功能将无法正常使用。

发明内容

本发明的目的在于提供一种基于MQ的智能限流方法，可按照不同的业务需求，将消息划分为不同的流量级别，并且可对不同流量级别的消息进行单独限流或者按照优先级进行限流，实现了对消息的有区别限流，避免了常规的限流方法对所有消息进行无区别限流可能引发系统无法正常运行等情况的发生。

为达此目的，本发明采用以下技术方案：

提供一种基于MQ的智能限流方法，具体步骤包括：

1）定义消息的流量级别；

2）创建用于限制对应流量级别的消息的至少一个流量限制器和用于对各所述流量限制器进行限流管理的流量限制管理器；

3）将业务产生的所述消息归类为对应的流量级别；

4）所述流量限制管理器根据需要限制的所述消息的流量级别，获取可限制该流量级别的所述消息的所述流量限制器；

5）所述流量限制器对达到限流条件的所述消息进行限流。

作为本发明的一种优选方案，所述消息的流量级别包括总流体消息流量级别、会议消息流量级别、Topic消息流量级别、客户端消息流量级别中的任意一种或多种。

作为本发明的一种优选方案，步骤3）中，将业务产生的所述消息归类为对应的流量级别的方法步骤为：

3.1）对所述消息进行内容解析，得到所述消息的流量级别标记；

3.2）将解析得到的所述流量级别标记与定义的流量级别进行匹配，根据匹配结果将所述消息归类为对应的流量级别。

作为本发明的一种优选方案，关联所述消息的所述流量级别标记包括所述消息的会议ID、Topic ID、客户端ID、消息ID中的任意一种或多种。

作为本发明的一种优选方案，步骤5）中，所述流量限制器对所述消息进行限流的步骤具体包括：

5.1）所述流量限制器在接收到所述流量限制请求后，判断窗口时间内消息数量是否超过一流量过载阈值，

若是，则对所述消息进行限流；

若否，则不限流，并在所述窗口时间内对所述消息进行计数；

5.2）一个所述窗口时间结束，所述流量限制器重置其消息计数状态，对业务产生的所述消息进行重新计数。

作为本发明的一种优选方案，当不同流量级别的所述消息同时达到限流条件时，所述流量限制管理器获取的各所述流量限制器按照自身的限流优先级由高到低按序对对应流量级别的所述消息进行限流。

作为本发明的一种优选方案，对于同一流量级别下的所述消息，所述流量限制器按照消息的限流优先级由高到低按序对所述消息进行限流。

本发明还提供了一种基于MQ的智能限流装置，所述装置包括：

消息流量级别定义模块，用于定义消息的流量级别；

流量限制器创建模块，用于创建用于限制对应流量级别的所述消息的流量限制器；

流量限制管理器创建模块，用于创建用于对各所述流量限制器进行限流管理的流量限制管理器；

消息归类模块，连接所述消息流量级别定义模块，用于将业务产生的所述消息归类为对应的流量级别；

流量限制器获取模块，分别连接所述流量限制器创建模块、所述流量限制管理器创建模块和所述消息归类模块，用于根据需要限制的所述消息对应的流量级别，并通过所述流量限制管理器获取可限制该流量级别的所述消息的所述流量限制器；

限流模块，连接所述流量限制器获取模块，用于通过获取的所述流量限制器对达到限流条件的所述消息进行限流。

作为本发明的一种优选方案，所述消息归类模块中具体包括：

消息内容解析单元，用于对所述消息进行内容解析，得到所述消息的流量级别标记；

流量归类单元，连接所述消息内容解析单元，用于将解析得到的所述流量级别标记与定义的流量级别进行匹配，根据匹配结果将所述消息归类为对应的流量级别。

作为本发明的一种优选方案，所述限流模块中具体包括：

流量过载阈值设置单元，用于提供给用户设置流量过载阈值；

窗口时间设置单元，用于提供给所述用户设置消息计数的窗口时间；

消息计数单元，连接所述窗口时间设置单元，用于在所述窗口时间内对所述消息进行计数；

是否限流判断单元，分别连接所述流量过载阈值设置单元和所述消息计数单元，用于判断所述窗口时间内累计的消息数量是否超过所述流量过载阈值，

若是，则判定所述消息达到限流条件；

若否，则判定所述消息未达到限流条件；

限流单元，连接所述是否限流判断单元，用于在达到限流条件后对所述消息进行限流；

消息计数重置单元，连接所述消息计数单元，用于在所述窗口时间期满后重置消息计数状态，对所述消息进行重新计数。

作为本发明的一种优选方案，所述限流模块中还包括：

第一限流优先级设置单元，用于提供给所述用户设置各所述流量限制器的限流优先级；

第二限流优先级设置单元，用于提供给所述用户设置不同类型的所述消息的限流优先级；

所述限流单元，还连接所述第一限流优先级设置单元和所述第二限流优先级设置单元，用于在不同流量级别的所述消息同时达到限流条件时，按照各所述流量限制器的限流优先级由高到低按序对对应流量级别的所述消息进行限流；

并用于对达到限流条件的同一流量级别下的各所述消息，按照消息自身的限流优先级由高到低按序对所述消息进行限流。

本发明还提供了一种计算机可读存储介质，包括执行指令，当电子设备的处理器执行所述执行指令时，所述处理器执行所述的基于MQ的智能限流方法。

本发明可按照不同的业务需求，将消息划分为不同的流量级别，并且可对不同流量级别的消息进行单独限流或者按照优先级进行限流，实现了消息限流的智能化，避免了常规的限流方法对所有消息进行无区别限流可能引发系统无法正常运行等情况的发生。

附图说明

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

图1是本发明一实施例提供的基于MQ的智能限流方法的步骤图；

图2是将业务产生的消息归类为对应的流量级别的方法步骤图；

图3是流量限制器对消息进行限流的方法步骤图；

图4是本发明一实施例提供的基于MQ的智能限流装置的结构示意图；

图5是智能限流装置中的消息归类模块的结构示意图；

图6是智能限流装置中的限流模块的结构示意图；

图7是mq-proxy连接在平台业务服务组件和MQ消息队列引擎间的示意图；

图8是流量限制管理器与各流量限制器的连接关系示意图；

图9是流量限制器判断是否达到限流条件的流程图；

图10是现有的令牌桶算法进行限流的原理图。

具体实施方式

下面结合附图并通过具体实施方式来进一步说明本发明的技术方案。

其中，附图仅用于示例性说明，表示的仅是示意图，而非实物图，不能理解为对本专利的限制；为了更好地说明本发明的实施例，附图某些部件会有省略、放大或缩小，并不代表实际产品的尺寸；对本领域技术人员来说，附图中某些公知结构及其说明可能省略是可以理解的。

本发明实施例的附图中相同或相似的标号对应相同或相似的部件；在本发明的描述中，需要理解的是，若出现术语“上”、“下”、“左”、“右”、“内”、“外”等指示的方位或位置关系为基于附图所示的方位或位置关系，仅是为了便于描述本发明和简化描述，而不是指示或暗示所指的装置或元件必须具有特定的方位、以特定的方位构造和操作，因此附图中描述位置关系的用语仅用于示例性说明，不能理解为对本专利的限制，对于本领域的普通技术人员而言，可以根据具体情况理解上述术语的具体含义。

在本发明的描述中，除非另有明确的规定和限定，若出现术语“连接”等指示部件之间的连接关系，该术语应做广义理解，例如，可以是固定连接，也可以是可拆卸连接，或成一体；可以是机械连接，也可以是电连接；可以是直接相连，也可以通过中间媒介间接相连，可以是两个部件内部的连通或两个部件的相互作用关系。对于本领域的普通技术人员而言，可以具体情况理解上述术语在本发明中的具体含义。

云会议平台是一套分布式软件系统，其通过多种业务服务组件为客户提供音视频传输、文字聊天等业务服务。各个业务服务组件之间主要通过RocketMQ（MQ消息队列引擎中的一种）进行消息通讯。为了将C++构型的业务服务组件接入MQ消息队列引擎，本发明为此在MQ消息队列引擎前端专门开发了程序，并命名为“mq-proxy”。mq-proxy通过thrift RPC的方式实现与各个业务服务组件的通信连接（mq-proxy连接在平台业务服务组件和MQ消息队列引擎间的示意图请参照图7，图7中的英文“MQ cluster”表示MQ消息队列引擎集群），mq-proxy在各业务服务组件和MQ消息队列引擎之间承担消息生产者或消费者的角色。mq-proxy还有一个最为重要的功能，就是能够对业务产生的消息进行智能化限流。

本发明实施例以对云会议平台产生的消息进行限流为例，对基于MQ的智能限流方法进行具体阐述。如图1所示，本实施例提供的基于MQ的智能限流方法，具体步骤包括：

步骤1）定义消息的流量级别；云会议平台产生的消息主要有会议消息（指召开云会议时产生的消息，比如音视频服务产生的消息）、Topic消息（主题消息）、客户端消息（客户端产生的消息）、总流体消息（云会议平台在一持续时间段内产生的所有类型的消息）等；本发明将会议消息形成的消息流定义为会议级别消息流量；将Topic消息形成的消息流定义为Topic级别消息流量；将客户端消息形成的消息流定义为客户端级别消息流量；将总流体消息形成的消息流定义为总流体级别消息流量；

步骤2）创建用于限制对应流量级别的消息的至少一个流量限制器和用于对各流量限制器进行限流管理的流量限制管理器；流量限制管理器与各流量限制器的连接关系请参照图8。一个流量限制器用于限制一种流量级别的消息。本实施例中，流量限制器包括但不限于总流体流量限制器（用于对总流体消息进行限流）、会议消息流量限制器（用于对会议消息进行限流）、Topic消息流量限制器（用于对Topic消息进行限流）和客户端消息流量限制器（用于对客户端消息进行限流）；

步骤3）将业务产生的消息（比如云会议平台的音视频业务服务组件产生的消息）归类为对应的流量级别；具体归类方法如图2所示，包括：

步骤3.1）对消息进行内容解析，得到消息的流量级别标记；比如，会议消息中包含有会议ID；Topic消息中含有Topic ID，客户端消息中含有客户端ID；总流体消息中含有消息ID，这些从消息头部中解析出的ID作为消息的流量级别标记；

步骤3.2）将解析得到的流量级别标记与定义的流量级别进行匹配，根据匹配结果将消息归类为对应的流量级别。比如会议ID与会议级别消息流量具有匹配关系，当消息中解析出会议ID，则确认该消息为会议消息，并将多个会议消息形成的消息流归类为会议级别消息流量。

步骤4）流量限制管理器根据需要限流的消息的流量级别，获取可限制该流量级别的消息的流量限制器；比如需要限制的为会议级别消息流量，则流量限制管理器获取会议消息流量限制器对会议消息进行限流；

步骤5）流量限制器对达到限流条件的消息进行限流。图3示出了流量限制器对消息进行限流的方法步骤图；图9示出了流量限制器判断是否达到限流条件的流程图，如图3和图9所示，流量限制器对消息进行限流的具体步骤包括：

步骤5.1）流量限制器在接收到流量限制请求后（由mq-proxy向流量限制器发送tryAcquire请求），判断窗口时间内消息数量是否超过一流量过载阈值，

若是，则对消息进行限流；

若否，则不限流，并在该窗口时间内对消息进行计数；

步骤5.2）一个窗口时间结束，流量限制器重置其消息计数状态，对业务产生的消息进行重新计数。

为了能够对不同流量级别的消息进行有区别的限流，优选地，当不同流量级别的消息同时达到限流条件时，流量限制管理器获取的各流量限制器按照自身的限流优先级由高到低按序对对应流量级别的消息进行限流。比如会议级别消息流量和客户端级别消息流量同时达到限流条件，但会议消息流量限制器的限流优先级高于客户端消息流量限制器的限流优先级，则会议消息流量限制器优先对会议消息进行限流，然后客户端消息流量限制器再对客户端消息进行限流。

进一步地，为了能够对同一流量级别下的消息进行有区别的限流，流量限制器对于同一流量级别下的消息，按照消息的限流优先级由高到低按序对消息进行限流。比如，总流体消息中包含会议消息、客户端消息和Topic消息，会议消息、客户端消息和Topic消息的限流优先级由高到低，当总流体消息达到限流条件时，总流体消息流量限制器优先对对总流体消息中的会议消息进行限流，然后再对客户端消息进行限流，最后再对Topic消息进行限流。

综上，本发明可按照不同的业务需求，将消息划分为不同的流量级别，并且可对不同流量级别的消息进行单独限流或者按照优先级进行限流，实现了对消息的有区别限流，避免了常规的限流方法对所有消息进行无区别限流可能引发系统无法正常运行等情况的发生。

本发明一实施例还提供了一种基于MQ的智能限流装置，如图4所示，该装置包括：

消息流量级别定义模块，用于定义消息的流量级别；消息的流量级别定义方法在上述的智能限流方法中已作了阐述，在此不再赘述；

流量限制器创建模块，用于创建用于限制对应流量级别的消息的流量限制器；

流量限制管理器创建模块，用于创建用于对各流量限制器进行限流管理的流量限制管理器；

消息归类模块，连接消息流量级别定义模块，用于将业务产生的消息归类为对应的流量级别；消息的归类方法在上述的智能限流方法中已作了阐述，在此不再赘述；

流量限制器获取模块，分别连接流量限制器创建模块、流量限制管理器创建模块和消息归类模块，用于根据需要限制的消息对应的流量级别，并通过流量限制管理器获取可限制该流量级别的消息的流量限制器；

限流模块，连接流量限制器获取模块，用于通过获取的流量限制器对达到限流条件的消息进行限流。具体的限流过程在上述的智能限流方法中已作了阐述，在此不再赘述。

如图4所示，消息归类模块中具体包括：

消息内容解析单元，用于对消息进行内容解析，得到消息的流量级别标记；

流量归类单元，连接消息内容解析单元，用于将解析得到的流量级别标记与定义的流量级别进行匹配，根据匹配结果将消息归类为对应的流量级别。

如图5所示，限流模块中具体包括：

流量过载阈值设置单元，用于提供给用户设置流量过载阈值；

窗口时间设置单元，用于提供给用户设置消息计数的窗口时间；

消息计数单元，连接窗口时间设置单元，用于在窗口时间内对消息进行计数；

是否限流判断单元，分别连接流量过载阈值设置单元和消息计数单元，用于根据窗口时间内累计的消息数量是否超过流量过载阈值，

若是，则判定消息达到限流条件；

若否，则判定消息未达到限流条件；

限流单元，连接是否限流判断单元，用于在达到限流条件后对消息进行限流；

消息计数重置单元，连接消息计数单元，用于在窗口时间期满后重置消息计数状态，对消息进行重新计数。

为了能够对不同流量级别的消息进行有区别的限流，优选地，限流模块中还包括：

第一限流优先级设置单元，用于提供给用户设置各流量限制器的限流优先级；

限流单元还连接第一限流优先级设置单元，用于在不同流量级别的消息同时达到限流条件时，按照各流量限制器的限流优先级由高到低按序对对应流量级别的消息进行限流。

为了能够对同一流量级别下的消息进行有区别的限流，更优选地，限流模块中还包括：

第二限流优先级设置单元，用于提供给用户设置不同类型的消息的限流优先级；

限流单元还连接第二限流优先级设置单元，用于对达到限流条件的同一流量级别下的各消息，按照消息自身的限流优先级由高到低按序对消息进行限流。

本发明还提供了一种计算机可读存储介质，包括执行指令，当电子设备的处理器执行该执行指令时，处理器执行上述的基于MQ的智能限流方法。

需要声明的是，上述具体实施方式仅仅为本发明的较佳实施例及所运用技术原理。本领域技术人员应该明白，还可以对本发明做各种修改、等同替换、变化等等。但是，这些变换只要未背离本发明的精神，都应在本发明的保护范围之内。另外，本申请说明书和权利要求书所使用的一些术语并不是限制，仅仅是为了便于描述。

Claims (12)

1.一种基于MQ的智能限流方法，其特征在于，具体步骤包括：

1）定义消息的流量级别；

2）创建用于限制对应流量级别的消息的至少一个流量限制器和用于对各所述流量限制器进行限流管理的流量限制管理器；

3）将业务产生的所述消息归类为对应的流量级别；

4）所述流量限制管理器根据需要限制的所述消息的流量级别，获取可限制该流量级别的所述消息的所述流量限制器；

5）所述流量限制器对达到限流条件的所述消息进行限流。

2.根据权利要求1所述的基于MQ的智能限流方法，其特征在于，所述消息的流量级别包括总流体消息流量级别、会议消息流量级别、Topic消息流量级别、客户端消息流量级别中的任意一种或多种。

3.根据权利要求1所述的基于MQ的智能限流方法，其特征在于，步骤3）中，将业务产生的所述消息归类为对应的流量级别的方法步骤为：

3.1）对所述消息进行内容解析，得到所述消息的流量级别标记；

3.2）将解析得到的所述流量级别标记与定义的流量级别进行匹配，根据匹配结果将所述消息归类为对应的流量级别。

4.根据权利要求3所述的基于MQ的智能限流方法，其特征在于，关联所述消息的所述流量级别标记包括所述消息的会议ID、Topic ID、客户端ID、消息ID中的任意一种或多种。

5.根据权利要求1所述的基于MQ的智能限流方法，其特征在于，步骤5）中，所述流量限制器对所述消息进行限流的步骤具体包括：

5.1）所述流量限制器在接收到所述流量限制请求后，判断窗口时间内消息数量是否超过一流量过载阈值，

若是，则对所述消息进行限流；

若否，则不限流，并在所述窗口时间内对所述消息进行计数；

5.2）一个所述窗口时间结束，所述流量限制器重置其消息计数状态，对业务产生的所述消息进行重新计数。

6.根据权利要求1所述的基于MQ的智能限流方法，其特征在于，当不同流量级别的所述消息同时达到限流条件时，所述流量限制管理器获取的各所述流量限制器按照自身的限流优先级由高到低按序对对应流量级别的所述消息进行限流。

7.根据权利要求1所述的基于MQ的智能限流方法，其特征在于，对于同一流量级别下的所述消息，所述流量限制器按照消息的限流优先级由高到低按序对所述消息进行限流。

8.一种基于MQ的智能限流装置，其特征在于，所述装置包括：

消息流量级别定义模块，用于定义消息的流量级别；

流量限制器创建模块，用于创建用于限制对应流量级别的所述消息的流量限制器；

流量限制管理器创建模块，用于创建用于对各所述流量限制器进行限流管理的流量限制管理器；

消息归类模块，连接所述消息流量级别定义模块，用于将业务产生的所述消息归类为对应的流量级别；

流量限制器获取模块，分别连接所述流量限制器创建模块、所述流量限制管理器创建模块和所述消息归类模块，用于根据需要限制的所述消息对应的流量级别，并通过所述流量限制管理器获取可限制该流量级别的所述消息的所述流量限制器；

限流模块，连接所述流量限制器获取模块，用于通过获取的所述流量限制器对达到限流条件的所述消息进行限流。

9.根据权利要求8所述的基于MQ的智能限流装置，其特征在于，所述消息归类模块中具体包括：

消息内容解析单元，用于对所述消息进行内容解析，得到所述消息的流量级别标记；

流量归类单元，连接所述消息内容解析单元，用于将解析得到的所述流量级别标记与定义的流量级别进行匹配，根据匹配结果将所述消息归类为对应的流量级别。

10.根据权利要求8所述的基于MQ的智能限流装置，其特征在于，所述限流模块中具体包括：

流量过载阈值设置单元，用于提供给用户设置流量过载阈值；

窗口时间设置单元，用于提供给所述用户设置消息计数的窗口时间；

消息计数单元，连接所述窗口时间设置单元，用于在所述窗口时间内对所述消息进行计数；

是否限流判断单元，分别连接所述流量过载阈值设置单元和所述消息计数单元，用于判断所述窗口时间内累计的消息数量是否超过所述流量过载阈值，

若是，则判定所述消息达到限流条件；

若否，则判定所述消息未达到限流条件；

限流单元，连接所述是否限流判断单元，用于在达到限流条件后对所述消息进行限流；

消息计数重置单元，连接所述消息计数单元，用于在所述窗口时间期满后重置消息计数状态，对所述消息进行重新计数。

11.根据权利要求10所述的基于MQ的智能限流装置，其特征在于，所述限流模块中还包括：

第一限流优先级设置单元，用于提供给所述用户设置各所述流量限制器的限流优先级；

第二限流优先级设置单元，用于提供给所述用户设置不同类型的所述消息的限流优先级；

所述限流单元，还连接所述第一限流优先级设置单元和所述第二限流优先级设置单元，用于在不同流量级别的所述消息同时达到限流条件时，按照各所述流量限制器的限流优先级由高到低按序对对应流量级别的所述消息进行限流；

并用于对达到限流条件的同一流量级别下的各所述消息，按照消息自身的限流优先级由高到低按序对所述消息进行限流。

12.一种计算机可读存储介质，包括执行指令，其特征在于，当电子设备的处理器执行所述执行指令时，所述处理器执行如权利要求1-7任意一项所述的基于MQ的智能限流方法。

Images