消息队列的动态再平衡调度方法与流程

1.本技术涉及数据处理技术领域，具体涉及一种消息队列的动态再平衡调度方法。


背景技术：

2.在消息处理过程中，由于话单类型不同导致业务逻辑复杂度不同，当指定业务处理的队列(queue)中的逻辑复杂高的话单比较多时，容易导致队列特定节点由于消费不及时反压导致的阻塞积压。现有技术通常采用线性自动或手工扩充消费者(consumer)的方式来应对积压，通过扩充消费者促使积压队列获得独立消费通道来加快消息处理。但是，这种场景需要主题(toptic)下的消息队列(messagequeue)配置的足够多，要有足够的资源提供扩展，而且只有匹配到积压消息队列的消费者才进行积压处理，其余大部分消费者通常处于空转状态，容易造成资源浪费。


技术实现要素：

3.本技术实施例提供一种消息队列的动态再平衡调度方法，用以解决现有技术在解决消息积压问题时容易造成资源浪费的技术问题。
4.第一方面，本技术实施例提供一种消息队列的动态再平衡调度方法，包括：
5.监控消息队列的消息处理情况，得到队列积压偏移量；
6.根据所述队列积压偏移量，判定待再平衡调度消息队列；
7.根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列。
8.在一个实施例中，所述根据所述队列积压偏移量，判定待再平衡调度消息队列，包括：
9.根据所述队列积压偏移量，结合孤立森林算法和再平衡触发阈值，得到异常积压消息队列；
10.根据时间段影响级别系数和主题影响级别系数，判定所述异常积压消息队列为待再平衡调度消息队列。
11.在一个实施例中，所述根据所述队列积压偏移量，结合孤立森林算法和再平衡触发阈值，得到异常积压消息队列，包括：
12.根据所述队列积压偏移量，通过所述孤立森林算法得到队列异常积压分数；
13.当所述队列异常积压分数接近1且大于所述再平衡触发阈值时，所述队列异常积压分数对应的消息队列为所述异常积压消息队列。
14.在一个实施例中，所述根据时间段影响级别系数和主题影响级别系数，判定所述异常积压消息队列为待再平衡调度消息队列，包括：
15.根据所述时间段影响级别系数和所述主题影响级别系数，得到所述异常积压消息队列的队列异常分数；
16.当所述异常积压消息队列的队列异常分数大于队列异常分数阈值时，判定所述异
常积压消息队列为待再平衡调度消息队列。
17.在一个实施例中，在所述当所述异常积压消息队列的队列异常分数大于队列异常分数阈值时，判定所述异常积压消息队列为待再平衡调度消息队列之后，还包括：
18.当所述待再平衡调度消息队列具有若干个时，根据若干所述待再平衡调度消息队列的队列异常分数设置若干所述待再平衡调度消息队列进行再平衡调度的优先级。
19.在一个实施例中，所述根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列，包括：
20.根据所述待再平衡调度消息队列的消息积压偏移量，得到预计回收偏移量范围；
21.根据所述预计回收偏移量范围，回收待回收消息；
22.通过异步批量消息发送方式将所述待回收消息重新分派至非积压消息队列。
23.在一个实施例中，在所述根据所述预计回收偏移量范围，回收待回收消息之后，还包括：
24.更新所述待再平衡调度消息队列的偏移量标志。
25.第二方面，本技术实施例提供一种消息队列的动态再平衡调度装置，包括：
26.消息队列监控模块，用于：监控消息队列的消息处理情况，得到队列积压偏移量；
27.待再平衡调度消息队列判定模块，用于：根据所述队列积压偏移量，判定待再平衡调度消息队列；
28.再平衡调度模块，用于：根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列。
29.第三方面，本技术实施例提供一种电子设备，包括处理器和存储有计算机程序的存储器，所述处理器执行所述程序时实现第一方面所述的消息队列的动态再平衡调度方法的步骤。
30.第四方面，本技术实施例提供一种计算机程序产品，包括计算机程序，所述计算机程序被处理器执行时实现第一方面所述的消息队列的动态再平衡调度方法的步骤。
31.本技术实施例提供的消息队列的动态再平衡调度方法，基于话单在时间序列上无特殊要求的特性，利用消息时间戳作为伸缩控制点，通过动态监控消息队列的消息处理情况，根据队列积压偏移量得到待再平衡调度消息队列，再根据待再平衡调度消息队列的消息积压偏移量得到积压的待回收消息，并将待回收消息重新分派至非积压消息队列，充分利用现有的空闲消费者，无需消耗更多的资源来扩充消费者，避免资源浪费以及增加系统运行复杂度，在资源消耗有限增加的情况下，显著提升积压消息队列的处理效率。
附图说明
32.为了更清楚地说明本技术或现有技术中的技术方案，下面将对实施例或现有技术描述中所需要使用的附图作一简单地介绍，显而易见地，下面描述中的附图是本技术的一些实施例，对于本领域普通技术人员来讲，在不付出创造性劳动的前提下，还可以根据这些附图获得其他的附图。
33.图1是本技术实施例提供的消息队列的动态再平衡调度方法的流程示意图；
34.图2是本技术实施例提供的消息队列的动态再平衡调度装置的结构示意图；
35.图3是本技术实施例提供的电子设备的结构示意图。
具体实施方式
36.为使本技术的目的、技术方案和优点更加清楚，下面将结合本技术实施例中的附图，对本技术中的技术方案进行清楚、完整地描述，显然，所描述的实施例是本技术一部分实施例，而不是全部的实施例。基于本技术中的实施例，本领域普通技术人员在没有作出创造性劳动前提下所获得的所有其他实施例，都属于本技术保护的范围。
37.图1为本技术实施例提供一种消息队列的动态再平衡调度方法的流程示意图。
38.参照图1，本技术实施例提供一种消息队列的动态再平衡调度方法，可以包括：
39.s110、监控消息队列的消息处理情况，得到队列积压偏移量。
40.s120、根据所述队列积压偏移量，判定待再平衡调度消息队列。
41.s130、根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列。
42.需要说明的是，本发明提供的一种消息队列的动态再平衡调度方法的执行主体可以是任何网络侧设备，例如消息队列管理系统等等。
43.需要说明的是，一般的消息处理流程包括：
44.由生产者(producer group)定时扫描消息队列服务器(broker)在路由中心(nameserver)上注册的主题(topic)配置信息，采取指定或者轮询方式向主题发送消息；
45.由消息队列服务器对消息队列进行存储；
46.由消费者(consumer group)根据路由中心上注册的主题配置信息，从指定消息队列中获取消息进行消费。
47.其中，生产者(producer group)用于向消息队列服务器(broker)队列发送话单消息；消息队列服务器(broker)用于接收并存储应用发送的话单消息，分片后存在主节点、备节点；消费者(consumer group)用于从消息队列服务器(broker)中获取消息给业务处理单元处理；路由中心(nameserver)用于注册并同步、发布生产者、消费者和消息队列服务器(broker)的信息；消息偏移量指的是：消息队列是无限长的数组，一条消息下来下标就会涨1，下标就是offset(偏移量)，通过offset可以定位这条消息。
48.本发明提供的一种消息队列的动态再平衡调度方法在一般的消息处理流程的同时，同步进行，以实时调度消息队列中的消息分派，在保证消耗最少资源的同时提高处理积压消息的效率。
49.本技术实施例提供的消息队列的动态再平衡调度方法，基于话单在时间序列上无特殊要求的特性，利用消息时间戳作为伸缩控制点，通过动态监控消息队列的消息处理情况，根据队列积压偏移量得到待再平衡调度消息队列，再根据待再平衡调度消息队列的消息积压偏移量得到积压的待回收消息，并将待回收消息重新分派至非积压消息队列，充分利用现有的空闲消费者，无需消耗更多的资源来扩充消费者，避免资源浪费以及增加系统运行复杂度，在资源消耗有限增加的情况下，显著提升积压消息队列的处理效率。
50.在步骤s110中，网络侧设备会监控消息队列的消息处理情况，得到队列积压偏移量。
51.需要说明的是，队列积压偏移量指的是被监控的所有消息队列各自的消息量，即队列积压偏移量包括若干消息队列的消息量。
52.具体的，网络侧设备可以采集每个消息队列的消息量，定时查询主题配置信息，以
快速且准确地计算出每个队列的队列积压偏移量。
53.在步骤s120中，网络侧设备会根据所述队列积压偏移量，判定待再平衡调度消息队列。
54.具体的，步骤s120可以包括：
55.根据所述队列积压偏移量，结合孤立森林算法和再平衡触发阈值，得到异常积压消息队列；
56.根据时间段影响级别系数和主题影响级别系数，判定所述异常积压消息队列为待再平衡调度消息队列。
57.需要说明的是，所述根据所述队列积压偏移量，结合孤立森林算法和再平衡触发阈值，得到异常积压消息队列，包括：
58.根据所述队列积压偏移量，通过所述孤立森林算法得到队列异常积压分数；
59.当所述队列异常积压分数接近1且大于所述再平衡触发阈值时，所述队列异常积压分数对应的消息队列为所述异常积压消息队列。
60.需要说明的是，再平衡触发阈值(th)可以动态调整，例如，采用平均法定期对前24个周期主题每小时处理消息量进行统计分析，并为每一期的数值加上了权重，如第24期的权重为0.5，第23期的权重为0.5的平方，保证距离预测期越远的数据作用越小，同时保证根据业务量的增加，实时动态变更再平衡触发阈值。其中，平均法可以采用下式(1)实现：
61.ft＝(at-1+at-2+at-3+
…
+at-n)/n
ꢀꢀ
(1)；
62.其中，ft表示对下一周期主题每小时处理消息量的预测值，n表示移动平均的周期个数；at-x表示周期主题每小时处理消息量的前期实际值。
63.具体的，基于孤立森林算法判断是否存在异常积压消息队列，通过队列积压偏移量作为异常点检测的方法来判断某个消息队列内的数据是否过多，通常队列积压偏移量应该近似，异常数据只占总数据量的少量，异常数据的数据特征值与正常数据的数据特征值差别很大，可以很明显地被找到，因此，采用类孤立树的算法可以判定某个消息队列是否为异常积压消息队列。
64.假设每个消息队列当前的队列积压偏移量为x，所有主题队列积压样本数据为x＝{x1，
…
，xn}，特征的维度为d，随机选择一个特征q及其分割值p，递归地分割数据集x，构建数据树，在积压样本数据已经无法分割或者所有特征都相同的时候，对样本点的树长度及异常得分进行排序，样本点就是路径长度最短或者异常得分最高的点。
65.样本点x的路径长度h(x)为从孤立树的根节点到叶子节点经过的边的数量。样本异常得分定义如下式(2)：
66.s(x,n)＝2-e(h(x))c(n)
ꢀꢀ
(2)；
67.其中，s(x，n)表示队列异常积压分数，e(h(x))表示样本x在孤立树中的路径长度的期望，c(n)表示给定一个包含n个样本的数据集，树的平均路径长度。
68.c(n)的计算公式如下式(3)所示：
69.c(n)＝2h(n-1)-2(n-1)n
ꢀꢀ
(3)；
70.其中h(n)表示调和数，用于标准化样本x的路径长度h(x)。
71.当e(h(x))趋于0时，s趋于1，即x的异常分数接近1，且当x的异常分数大于再平衡触发阈值(th)时，准确判定队列异常积压分数对应的消息队列为异常积压消息队列，该消
息队列存在异常消息积压的情况，需要再平衡调度。
72.需要说明的是，所述根据时间段影响级别系数和主题影响级别系数，判定所述异常积压消息队列为待再平衡调度消息队列，包括：
73.根据所述时间段影响级别系数和所述主题影响级别系数，得到所述异常积压消息队列的队列异常分数；
74.当所述异常积压消息队列的队列异常分数大于队列异常分数阈值时，判定所述异常积压消息队列为待再平衡调度消息队列。
75.需要说明的是，时间段影响级别系数(h)可以根据历史数据中一天内不同时间段的消息处理量来为一天中每个时间段设置对应的时间段影响级别系数，例如可能在历史数据中每一天的下午14时至17时消息处理量的平均值是一天中最大的，那么下午14时至17时的时间段影响级别系数可以设置为最高等等。主题影响级别系数可以根据实际情况来预先设置。网络侧设备可以根据异常积压消息队列的时间段影响级别系数和主题影响级别系数来得到队列异常分数，当异常积压消息队列的队列异常分数大于队列异常分数阈值时，判定异常积压消息队列为待再平衡调度消息队列。
76.另外，还可以结合过去24个周期的主题小时消息处理量进行自动统计分析，按主题小时消息处理量均值进行冒泡排序对时间段影响级别系数进行排序。
77.需要说明的是，当所述待再平衡调度消息队列具有若干个时，终端侧可以根据若干所述待再平衡调度消息队列的队列异常分数设置若干所述待再平衡调度消息队列进行再平衡调度的优先级。
78.具体的，假如若干个异常积压消息队列均被判定为待再平衡调度消息队列，终端侧设备可以根据若干个待再平衡调度消息队列的队列异常分数进行高低排序，队列异常分数越高的待再平衡调度消息队列优先进行再平衡调度，以高效应对消息积压的问题。
79.在步骤s130中，网络侧设备会根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列。
80.需要说明的是，待再平衡调度消息队列的消息积压偏移量是指待再平衡调度消息队列本身的消息积压偏移量，即一个待再平衡调度消息队列的消息积压偏移量。
81.具体的，步骤s130可以包括：
82.根据所述待再平衡调度消息队列的消息积压偏移量，得到预计回收偏移量范围；
83.根据所述预计回收偏移量范围，回收待回收消息；
84.通过异步批量消息发送方式将所述待回收消息重新分派至非积压消息队列。
85.例如，网络侧设备可以先得到待再平衡调度消息队列的队列名称，例如可以从策略调度中心标志池中得到待再平衡调度消息队列的队列名称以及消息积压偏移量，然后根据待再平衡调度消息队列的队列名称和消息积压偏移量(offset值)，计算预计回收偏移量范围，预计回收偏移量范围可以表示为：(offset+1至sum(delay)
×
(n-1)/n)，然后根据预计回收偏移量范围，回收预计回收偏移量范围内的待回收消息，再通过异步批量消息发送方式将待回收消息重新分派至非积压消息队列，即对待回收消息，多条消息一次性发送到api调用接口并立即返回，这种消息发送方式能够避免消息堵塞，进一步提高解决消息积压问题的速度，以及将待回收消息重新分派至非积压消息队列时可以将待回收消息平均分派至每一个非积压消息队列，或者也可以根据非积压消息队列的实际情况来分配待回收消
息，以实现在最短时间内最快地处理积压消息。
86.需要说明的是，在根据预计回收偏移量范围，回收待回收消息之后，终端侧设备可以更新待再平衡调度消息队列的偏移量标志。例如，将待再平衡调度消息队列的偏移量标志修改为：(sum(delay)
×
(n-1)/n+1)，用于通知原有消费者跳过预计回收偏移量范围内的待回收消息，继续正常处理其余消息。
87.下表示出通过现有技术与通过本技术实施例提供的消息队列的动态再平衡调度方法处理积压消息问题的效果对比：
[0088][0089]
可见，通过本技术实施例提供的消息队列的动态再平衡调度方法处理积压消息问题的网络侧设备的性能处理提升约18.19倍，具有显著的效果，能够充分利用资源。
[0090]
更多地，在更多的试运行中发现，在现有资源基础上，单地市日处理22亿条话单，在144个队列、28个消费者的配置下，4台业务处理服务器(运行本技术实施例提供的消息队列的动态再平衡调度方法)的高峰时段cpu平均使用率从90％降低到50％；经3个月试运行，因话单积压导致的人工介入从每月平均27次降为平均每月2次。
[0091]
本技术实施例提供的消息队列的动态再平衡调度方法，基于话单在时间序列上无特殊要求的特性，利用消息时间戳作为伸缩控制点，通过孤立森林算法，结合再平衡触发阈值、时间段影响级别系数、以及主题影响级别系数，对待再平衡调度消息队列建立可控范围的消息回收重分派机制，将积压消息队列中的待回收消息转移至相对空闲的非积压消息队列，利用现有的空闲消费者，避免消费者空转浪费资源，在资源消耗有限增加的情况下，实现显著提升积压队列的处理速度。
[0092]
另外，本技术实施例提供的消息队列的动态再平衡调度方法可以延伸到类似消息队列等消息中间件队列积压处理方法中，具有一定的扩展性。
[0093]
下面对本技术实施例提供的消息队列的动态再平衡调度装置进行描述，下文描述的消息队列的动态再平衡调度装置与上文描述的消息队列的动态再平衡调度方法可相互对应参照。
[0094]
图2为本技术实施例提供一种消息队列的动态再平衡调度装置的结构示意图。
[0095]
参照图2，本技术实施例提供一种消息队列的动态再平衡调度装置，可以包括：
[0096]
消息队列监控模块210，用于：监控消息队列的消息处理情况，得到队列积压偏移量；
[0097]
待再平衡调度消息队列判定模块220，用于：根据所述队列积压偏移量，判定待再
平衡调度消息队列；
[0098]
再平衡调度模块230，用于：根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列。
[0099]
在一种实施例中，所述待再平衡调度消息队列判定模块220可以包括：
[0100]
异常积压消息队列得到子模块，用于：根据所述队列积压偏移量，结合孤立森林算法和再平衡触发阈值，得到异常积压消息队列；
[0101]
待再平衡调度消息队列判定子模块，用于：根据时间段影响级别系数和主题影响级别系数，判定所述异常积压消息队列为待再平衡调度消息队列。
[0102]
在一种实施例中，所述异常积压消息队列得到子模块可以包括：
[0103]
队列异常积压分数得到子模块，用于：根据所述队列积压偏移量，通过所述孤立森林算法得到队列异常积压分数；
[0104]
异常积压消息队列判定子模块，用于：当所述队列异常积压分数接近1且大于所述再平衡触发阈值时，所述队列异常积压分数对应的消息队列为所述异常积压消息队列。
[0105]
在一种实施例中，所述待再平衡调度消息队列判定子模块可以包括：
[0106]
队列异常分数得到子模块，用于：根据所述时间段影响级别系数和所述主题影响级别系数，得到所述异常积压消息队列的队列异常分数；
[0107]
待再平衡调度消息队列得到子模块，用于：当所述异常积压消息队列的队列异常分数大于队列异常分数阈值时，判定所述异常积压消息队列为待再平衡调度消息队列。
[0108]
在一种实施例中，还包括：
[0109]
平衡调度优先级设置模块，用于：当所述待再平衡调度消息队列具有若干个时，根据若干所述待再平衡调度消息队列的队列异常分数设置若干所述待再平衡调度消息队列进行再平衡调度的优先级。
[0110]
在一种实施例中，所述再平衡调度模块230可以包括：
[0111]
预计回收偏移量范围得到子模块，用于：根据所述待再平衡调度消息队列的消息积压偏移量，得到预计回收偏移量范围；
[0112]
待回收消息得到子模块，用于：根据所述预计回收偏移量范围，回收待回收消息；
[0113]
待回收消息分派子模块，用于：通过异步批量消息发送方式将所述待回收消息重新分派至非积压消息队列。
[0114]
在一种实施例中，还包括：
[0115]
偏移量标志更新模块，用于：更新所述待再平衡调度消息队列的偏移量标志。
[0116]
图3示例了一种电子设备的实体结构示意图，如图3所示，该电子设备可以包括：处理器(processor)810、通信接口(communication interface)820、存储器(memory)830和通信总线840，其中，处理器810，通信接口820，存储器830通过通信总线840完成相互间的通信。处理器810可以调用存储器830中的计算机程序，以执行消息队列的动态再平衡调度方法的步骤，例如包括：
[0117]
监控消息队列的消息处理情况，得到队列积压偏移量；
[0118]
根据所述队列积压偏移量，判定待再平衡调度消息队列；
[0119]
根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列。
[0120]
此外，上述的存储器830中的逻辑指令可以通过软件功能单元的形式实现并作为独立的产品销售或使用时，可以存储在一个计算机可读取存储介质中。基于这样的理解，本技术的技术方案本质上或者说对现有技术做出贡献的部分或者该技术方案的部分可以以软件产品的形式体现出来，该计算机软件产品存储在一个存储介质中，包括若干指令用以使得一台计算机设备(可以是个人计算机，服务器，或者网络设备等)执行本技术各个实施例所述方法的全部或部分步骤。而前述的存储介质包括：u盘、移动硬盘、只读存储器(rom，read-only memory)、随机存取存储器(ram，random access memory)、磁碟或者光盘等各种可以存储程序代码的介质。
[0121]
另一方面，本技术实施例还提供一种计算机程序产品，所述计算机程序产品包括计算机程序，所述计算机程序可存储在非暂态计算机可读存储介质上，所述计算机程序被处理器执行时，计算机能够执行上述各实施例所提供的消息队列的动态再平衡调度方法的步骤，例如包括：
[0122]
监控消息队列的消息处理情况，得到队列积压偏移量；
[0123]
根据所述队列积压偏移量，判定待再平衡调度消息队列；
[0124]
根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列。
[0125]
另一方面，本技术实施例还提供一种处理器可读存储介质，所述处理器可读存储介质存储有计算机程序，所述计算机程序用于使处理器执行上述各实施例所提供的消息队列的动态再平衡调度方法的步骤，例如包括：
[0126]
监控消息队列的消息处理情况，得到队列积压偏移量；
[0127]
根据所述队列积压偏移量，判定待再平衡调度消息队列；
[0128]
根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列。
[0129]
所述处理器可读存储介质可以是处理器能够存取的任何可用介质或数据存储设备，包括但不限于磁性存储器(例如软盘、硬盘、磁带、磁光盘(mo)等)、光学存储器(例如cd、dvd、bd、hvd等)、以及半导体存储器(例如rom、eprom、eeprom、非易失性存储器(nand flash)、固态硬盘(ssd))等。
[0130]
以上所描述的装置实施例仅仅是示意性的，其中所述作为分离部件说明的单元可以是或者也可以不是物理上分开的，作为单元显示的部件可以是或者也可以不是物理单元，即可以位于一个地方，或者也可以分布到多个网络单元上。可以根据实际的需要选择其中的部分或者全部模块来实现本实施例方案的目的。本领域普通技术人员在不付出创造性的劳动的情况下，即可以理解并实施。
[0131]
通过以上的实施方式的描述，本领域的技术人员可以清楚地了解到各实施方式可借助软件加必需的通用硬件平台的方式来实现，当然也可以通过硬件。基于这样的理解，上述技术方案本质上或者说对现有技术做出贡献的部分可以以软件产品的形式体现出来，该计算机软件产品可以存储在计算机可读存储介质中，如rom/ram、磁碟、光盘等，包括若干指令用以使得一台计算机设备(可以是个人计算机，服务器，或者网络设备等)执行各个实施例或者实施例的某些部分所述的方法。
[0132]
最后应说明的是：以上实施例仅用以说明本技术的技术方案，而非对其限制；尽管
参照前述实施例对本技术进行了详细的说明，本领域的普通技术人员应当理解：其依然可以对前述各实施例所记载的技术方案进行修改，或者对其中部分技术特征进行等同替换；而这些修改或者替换，并不使相应技术方案的本质脱离本技术各实施例技术方案的精神和范围。

技术特征：
1.一种消息队列的动态再平衡调度方法，其特征在于，包括：监控消息队列的消息处理情况，得到队列积压偏移量；根据所述队列积压偏移量，判定待再平衡调度消息队列；根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列。2.根据权利要求1所述的消息队列的动态再平衡调度方法，其特征在于，所述根据所述队列积压偏移量，判定待再平衡调度消息队列，包括：根据所述队列积压偏移量，结合孤立森林算法和再平衡触发阈值，得到异常积压消息队列；根据时间段影响级别系数和主题影响级别系数，判定所述异常积压消息队列为待再平衡调度消息队列。3.根据权利要求2所述的消息队列的动态再平衡调度方法，其特征在于，所述根据所述队列积压偏移量，结合孤立森林算法和再平衡触发阈值，得到异常积压消息队列，包括：根据所述队列积压偏移量，通过所述孤立森林算法得到队列异常积压分数；当所述队列异常积压分数接近1且大于所述再平衡触发阈值时，判定所述队列异常积压分数对应的消息队列为所述异常积压消息队列。4.根据权利要求3所述的消息队列的动态再平衡调度方法，其特征在于，所述根据时间段影响级别系数和主题影响级别系数，判定所述异常积压消息队列为待再平衡调度消息队列，包括：根据所述时间段影响级别系数和所述主题影响级别系数，得到所述异常积压消息队列的队列异常分数；当所述异常积压消息队列的队列异常分数大于队列异常分数阈值时，判定所述异常积压消息队列为待再平衡调度消息队列。5.根据权利要求4所述的消息队列的动态再平衡调度方法，其特征在于，在所述当所述异常积压消息队列的队列异常分数大于队列异常分数阈值时，判定所述异常积压消息队列为待再平衡调度消息队列之后，还包括：当所述待再平衡调度消息队列具有若干个时，根据若干所述待再平衡调度消息队列的队列异常分数设置若干所述待再平衡调度消息队列进行再平衡调度的优先级。6.根据权利要求1-5任一项所述的消息队列的动态再平衡调度方法，其特征在于，所述根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列，包括：根据所述待再平衡调度消息队列的消息积压偏移量，得到预计回收偏移量范围；根据所述预计回收偏移量范围，回收待回收消息；通过异步批量消息发送方式将所述待回收消息重新分派至非积压消息队列。7.根据权利要求6所述的消息队列的动态再平衡调度方法，其特征在于，在所述根据所述预计回收偏移量范围，回收待回收消息之后，还包括：更新所述待再平衡调度消息队列的偏移量标志。8.一种消息队列的动态再平衡调度装置，其特征在于，包括：消息队列监控模块，用于：监控消息队列的消息处理情况，得到队列积压偏移量；
待再平衡调度消息队列判定模块，用于：根据所述队列积压偏移量，判定待再平衡调度消息队列；再平衡调度模块，用于：根据所述待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将所述待回收消息重新分派至非积压消息队列。9.一种电子设备，包括处理器和存储有计算机程序的存储器，其特征在于，所述处理器执行所述计算机程序时实现权利要求1至7任一项所述的消息队列的动态再平衡调度方法的步骤。10.一种计算机程序产品，包括计算机程序，其特征在于，所述计算机程序被处理器执行时实现权利要求1至7任一项所述的消息队列的动态再平衡调度方法的步骤。

技术总结
本申请提供消息队列的动态再平衡调度方法，涉及数据处理技术领域。所述方法包括：监控消息队列的消息处理情况，得到队列积压偏移量；根据队列积压偏移量，得到待再平衡调度消息队列；根据待再平衡调度消息队列的消息积压偏移量，得到待回收消息，并将待回收消息重新分派至非积压消息队列。本申请提供的消息队列的动态再平衡调度方法，基于话单在时间序列上无特殊要求的特性，利用消息时间戳作为伸缩控制点，通过动态监控消息队列的消息处理情况，得到积压的待回收消息并将其重派至非积压消息队列，充分利用现有的空闲消费者，无需消耗更多的资源来扩充消费者，避免资源浪费以及增加系统运行复杂度，同时显著提升积压消息队列的处理效率。的处理效率。的处理效率。


技术研发人员：王娟 汤磊 张蔷 单浩
受保护的技术使用者：中国移动通信集团有限公司
技术研发日：2022.01.27
技术公布日：2023/8/8