消息队列中的排队理论

1，单队列单消费者

• MQ中默认一个队列只对应一个消费者，假设单位时间内新来$\lambda$$\lambda$个消息，消费者单位时间可以处理$\mu$$\mu$条消息，不能及时被消费的消息将缓存在队列中，上述系统可以建模为一个M/M/1系统¹。

• 队列中消息的到达和消费可以建模为生灭过程²，假设有$k$$k$条待处理消息的状态为$S_k$$S_k$，处于该状态的概率为$P_k$$P_k$。

  

• $S_k$$S_k$可能因为有新消息到达转移到$S_{k+1}$$S_{k+1}$，或者现有消息被消费转移到$S_{k-1}$$S_{k-1}$，他们的转移速率可以定义为$P_k\lambda ,P_k\mu$$P_k\lambda, P_{k}\mu$，二者之和定义为流出状态$S_k$$S_k$的速率。

• $S_k$$S_k$也可以通过在$S_{k-1}$$S_{k-1}$时有新消息到达，或者在状态$S_{k+1}$$S_{k+1}$时现有消息被消费转移得到。他们转移的速率可以定义为$P_{k-1}\lambda ,P_{k+1}\mu$$P_{k-1}\lambda, P_{k+1}\mu$，二者之和定义为流入状态$S_k$$S_k$的速率。

• 如果队列在状态$S_k$$S_k$达到平衡，则流入状态$S_k$$S_k$的速率等于流出状态$S_k$$S_k$的速率，

  对于$k=0$$k=0$时有

  $$\begin{array}{}\text{(1)}& P_0\lambda =P_1\mu \end{array}$$

  对于$k>0$$k>0$有

  $$\begin{array}{}\text{(2)}& P_k\lambda +P_k\mu =P_{k-1}\lambda +P_{k+1}\mu \end{array}$$

  结合上述两式可以获得

  $$\begin{array}{}\text{(3)}& P_k=P_{k-1}\frac{\lambda }{\mu },k>0\end{array}$$

  队列全部状态的概率和为一$\sum _{k=0}^{\mathrm{\infty }}{P}_k=1$$\sum_{k=0}^{\infty}P_k=1$，通过对$P_k$$P_k$进行等比数列求和求解得到

  $$\begin{array}{}\text{(4)}& P_k=(1-\frac{\lambda }{\mu })(\frac{\lambda }{\mu }{)}^k\end{array}$$

• 系统消息总数均值为处于各个状态的概率和：$L=\sum _{k=0}^{\mathrm{\infty }}k{P}_k=\sum _{k=0}^{\mathrm{\infty }}k(1-\frac{\lambda }{\mu })(\frac{\lambda }{\mu }{)}^k=\frac{\lambda }{\mu -\lambda }$$L=\sum_{k=0}^\infty kP_k=\sum_{k=0}^\infty k(1-\frac{\lambda}{\mu})(\frac{\lambda}{\mu})^k=\frac{\lambda}{\mu-\lambda}$

• 等待消息总数均值为系统消息总数均值减去正在处理的消息数：$L_q=L-\frac{\lambda }{\mu }=\frac{\lambda }{\mu -\lambda }-\frac{\lambda }{\mu }$$L_q=L-\frac{\lambda}{\mu}=\frac{\lambda}{\mu-\lambda}-\frac{\lambda}{\mu}$

• 应用Little Law³获得消息在系统的平均逗留时间： $W=\frac{L}{\lambda }=\frac{1}{\mu -\lambda }$$W=\frac{L}{\lambda }=\frac{1}{\mu−\lambda}$

• 应用Little Law获得消息在队列中平均等待时间：${𝑊}_q=\frac{L_{𝑞}}{\lambda }=\frac{\lambda }{\mu (\mu -\lambda )}$$𝑊_q=\frac{L_𝑞}{\lambda}=\frac{\lambda}{\mu(\mu-\lambda)}$

2，单队列多消费者

• MQ开启批量消费，一个队列对应n个消费者，多个消费者共享同一个队列，假设单位时间内新到达$\lambda$$\lambda$个消息，每个消费者单位时间最多可以处理$\mu$$\mu$条消息，不能及时被消费的消息将缓存在队列中，上述系统可以建模为一个M/M/n系统⁴。

  

• $S_k$$S_k$可能因为有新消息到达从而转移到$S_{k+1}$$S_{k+1}$，或者现有消息被消费转移到状态$S_{k-1}$$S_{k-1}$。

  如果队列消息数小于消费者数$k<n$$k<n$，则他们转移的速率可以分别定义为$P_k\lambda ,k{P}_k\mu$$P_k\lambda, kP_{k}\mu$；如果队列消息数大于等于消费者数$k>=n$$k>=n$，则他们转移的速率可以定义为$P_k\lambda ,n{P}_k\mu$$P_k\lambda, nP_{k}\mu$，二者之和定义为流出状态$S_k$$S_k$的速率。

• $S_k$$S_k$可以通过在$S_{k-1}$$S_{k-1}$时有新消息到达，或者在状态$S_{k+1}$$S_{k+1}$时现有消息被消费转移得到。

  如果队列消息数小于消费者$k<n$$k<n$，则他们转移的速率可以定义为$P_{k-1}\lambda ,(k+1)P_{k+1}\mu$$P_{k-1}\lambda, (k+1)P_{k+1}\mu$；如果队列消息数大于等于消费者数$k>=n$$k>=n$，则他们转移的速率可以定义为$P_{k-1}\lambda ,n{P}_{k+1}\mu$$P_{k-1}\lambda, nP_{k+1}\mu$，二者之和定义为流入状态$S_k$$S_k$的速率。

• 如果队列在状态$S_k$$S_k$时到达平衡，则流入状态$S_k$$S_k$的速率等于流出状态$S_k$$S_k$的速率，

  对于$k=0$$k=0$有

  $$\begin{array}{}\text{(5)}& P_0\lambda =P_1\mu \end{array}$$

  对于$k<n$$k<n$有

  $$\begin{array}{}\text{(6)}& P_k\lambda +k{P}_k\mu =P_{k-1}\lambda +(k+1)P_{k+1}\mu \end{array}$$

  对于$k\ge n$$k\geq n$有

  $$\begin{array}{}\text{(7)}& P_k\lambda +n{P}_k\mu =P_{k-1}\lambda +n{P}_{k+1}\mu \end{array}$$

  综合上式递推得到系统空闲概率$P_0$$P_0$，以及系统包含$k$$k$条消息的概率为

  $$\begin{array}{}\text{(8)}& \begin{array}{l}{P}_0={(\sum _{k=0}^{n-1}\frac{(\lambda /\mu {)}^k}{k!}+\frac{(\lambda /\mu {)}^n}{n!\cdot (1-\rho )})}^{-1}\\ P_k=\frac{(\lambda /\mu {)}^k}{k!}{P}_0& k<n\\ P_k=\frac{(\lambda /\mu {)}^k}{n^{k-n}\cdot n!}{P}_0& k\ge n\end{array}\end{array}$$

• 在队列中等待处理的消息总数均值为：$L_q=\frac{(n\lambda /\mu {)}^n}{n!{(1-\frac{\lambda }{n\mu })}^2}{P}_0$$L_q = \frac{(n \lambda/\mu)^n}{n! \left(1 - \frac{\lambda}{n\mu}\right)^2} P_0$

• 系统平均消息总数均值为处于等待状态的消息总数均值加上正在处理的消息总数均值：$L=L_q+n\lambda /\mu$$L = L_q + n \lambda/\mu$

• 应用Little Law获得消息在系统的平均逗留时间均值： $W=\frac{L}{\lambda }$$W = \frac{L}{\lambda}$

• 应用Little Law获得消息在队列中平均等待均值时间：$W_q=\frac{L_q}{\lambda }$$W_q = \frac{L_q}{\lambda}$

3，性能对比

• 现在假设有5个消费者n=5，每个消费者单位时间内可以处理一条消息$\mu =1$$\mu=1$。对于单队列单消费者情形构建n个独立的队列，每个队列关联一个消费者；对于单队列多消费者情形，系统只有一个队列，n个消费者共享该队列。

• 对于n个单队列单消费者构成的系统，定义系统负载$\rho =\frac{n\lambda }{n\mu }=\frac{\lambda }{\mu }$$\rho=\frac{n\lambda}{n\mu}=\frac{\lambda}{\mu}$；

  对于包含n个消费者的单队列多消费者系统，定义系统负载$\rho =\frac{\lambda }{n\mu }$$\rho=\frac{\lambda}{n\mu}$。

  当二者系统负载相同时，将在单位时间内收到相同数量的新消息。下图展示了二者在相同系统负载时队列中等待处理的消息总数均值$L_q$$L_q$、系统消息总数均值$L$$L$、消息在系统逗留时间均值$W$$W$，和消息在队列中等待时间均值$W_q$$W_q$，随系统负载$\rho$$\rho$的变化关系。

• 可见当新消息到来速度小于系统消息消费能力$\rho <1$$\rho <1$的时候，M/M/n系统能立即处理新到来消息，没有消息积压，也没有等待处理耗时。然而单队列单消费者系统相较于单队列多消费者系统，消息积压较为严重，消息等待处理时间更长，在系统负载增大时，劣势更加明显。

4，性能优化

• MQ出于消息顺序性消费考虑、以及单个队列读写性能受限的原因，默认限制每个分区只能由一个消费者组中的一个消费者消费。

• M/M/1和M/M/n系统性能差距主要来源于：由于队列间独立，并且没有类似Golang中GMP的消息窃取机制（同样出于有序性要求），当一些队列面临高压力出现长队列，而其他队列可能暂时空闲，相应空闲队列对应的消费者空转，导致系统资源利用率低，消息不能及时处理。

• MQ的解决方式是使用再平衡机制，在消费者与队列之间动态地重新分配队列来实现负载平衡。再平衡机制将在周期性延迟、消费者发生变化，以及订阅topic变化后，被触发执行再平衡。

4.1，RocketMQ 再平衡实现

• Rocket MQ中在RebalanceService#run()中定时(默认20s)触发再平衡流程。

  1
  // RebalanceService#run()
  2
  public void run() {
  3
      long realWaitInterval = waitInterval;
  4
      while (!this.isStopped()) {
  5
          // 定时休眠后触发再平衡
  6
          this.waitForRunning(realWaitInterval);
  7
          boolean balanced = this.mqClientFactory.doRebalance();
  8
          }
  9
      }
  10
  }
  

  具体通过RebalanceImpl实施平衡：再平衡过程中，获取当前所有活跃的消费者列表和所有队列的列表，然后根据轮询或者一致性哈希分配等算法将topic下队列分配给所有活跃的消费者。

  xxxxxxxxxx
  9
  1
  // RebalanceImpl#doRebalance()
  2
  public boolean doRebalance(final boolean isOrder) {
  3
      // 获取当前所有活跃的消费者列表和所有队列的列表
  4
      for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
  5
          final String topic = entry.getKey();
  6
          // 执行再平衡
  7
          boolean result = this.rebalanceByTopic(topic, isOrder);
  8
      }
  9
  }
  

  xxxxxxxxxx
  16
  1
  // RebalanceImpl#rebalanceByTopic
  2
  private boolean rebalanceByTopic(final String topic, final boolean isOrder) {
  3
      // 获取topic对应的队列和consumer信息
  4
      Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
  5
      List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
  6
      // 选择再平衡策略
  7
      AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;
  8
      // 完成再平衡
  9
      List<MessageQueue> allocateResult = null;
  10
      allocateResult = strategy.allocate(
  11
          this.consumerGroup,
  12
          this.mQClientFactory.getClientId(),
  13
          mqAll,
  14
          cidAll);
  15
      }
  16
  }
  

• 再平衡策略决定队列和消费者间的对应关系，RocketMQ提供了下列分配策略

  • AllocateMessageQueueAveragely：默认策略，先计算队列数除以消费者数的余数$N_q\mathrm{\%}{N}_c=N_m$$N_q\%N_c=N_m$，然后将余数$N_m$$N_m$个队列依次分配给$N_m$$N_m$个消费者，最后把剩余的队列连续的分为$(N_q-N_m)/N_c$$(N_q-N_m)/N_c$份，依次分配给全部消费者。

  • AllocateMessageQueueAveragelyByCircle：每个消费者依次分配一个队列，循环分配，直至分配完毕全部队列。

  • AllocateMessageQueueConsistentHash：通过一致性hash算法计算队列所属消费者。此策略可以尽量减少 因为队列或者消费者数量发生变化，导致消费者与队列关联关系变化，从而需要重新建立TCP连接的数量。同时再平衡阶段消费者将不能正常消费消息，更少的关系变化，可以更快的完成再平衡过程，减少MQ不可用时间。

4.2，Kafka 再平衡实现

• 再平衡流程将在消费者组或者topic发生变化后，当前消费者组领导者加入消费组后进行

  xxxxxxxxxx
  13
  1
  // ConsumerCoordinator#onJoinComplete
  2
  protected void onJoinComplete(int generation, String memberId, String assignmentStrategy, ByteBuffer assignmentBuffer) {
  3
      // 再平衡策略
  4
      ConsumerPartitionAssignor assignor = this.lookupAssignor(assignmentStrategy);
  5
      // 待分配分区
  6
      SortedSet<TopicPartition> assignedPartitions = new TreeSet(COMPARATOR);
  7
      assignedPartitions.addAll(assignment.partitions());
  8
      // 执行分配
  9
      firstException.compareAndSet((Object)null, this.invokeOnAssignment(assignor, assignment));
  10
      this.subscriptions.assignFromSubscribed(assignedPartitions);
  11
      firstException.compareAndSet((Object)null,  this.invokePartitionsAssigned(addedPartitions));
  12
      }
  13
  }
  

• 再平衡策略：Kafka提供了和RocketMQ相似的分配策略

  • RangeAssignor与RocketMQ的AllocateMessageQueueAveragely策略一致。

  • RoundRobinAssignor与RocketMQ的AllocateMessageQueueAveragelyByCircle策略一致。

  • StickyAssignor：分配尽量均匀，尽量与上一次分配的相同，尽量减少分配关系的变动，从而减少需要重新建立消费者与分区之间TCP连接的数量，缩短再平衡过程耗时，降低MQ不可用时间。。

  • CooperativeStickyAssignor：两阶段再平衡策略，先尝试只移动最少量的分区完成再分配，如果分配后不满足平衡条件，再进行完整的再平衡。

附录

队列长度与逗留时间计算绘制脚本

x
1
from scipy.special import factorial
2
import numpy as np
3
import matplotlib.pyplot as plt
4

5

6
# MM1 队列模型
7
def mm1_queue(lambda_, mu):
8
    if lambda_ >= mu:
9
        return "系统不稳定"
10
    rho = lambda_ / mu
11
    # 等候+服务人数
12
    L = rho / (1 - rho)
13
    # 等候+服务时间
14
    W = 1 / (mu - lambda_)
15
    # 等候人数
16
    Lq = rho**2 / (1 - rho)
17
    # 等候时间
18
    Wq = rho / (mu - lambda_)
19
    return L, W, Lq, Wq
20

21
# MMn 队列模型
22
def mmn_queue(lambda_, mu, n):
23
    # 计算流量强度
24
    rho = lambda_ / (n * mu)
25
    if rho >= 1:
26
        return "系统不稳定，请确保 rho < 1"
27

28
    # 计算P0
29
    sum_p = sum((lambda_ / mu)**k / factorial(k) for k in range(n))
30
    p0 = (sum_p + (lambda_ / mu)**n / (factorial(n) * (1 - rho)))**(-1)
31
    # 计算队列中的平均客户数 Lq
32
    lq = (rho**n * rho / factorial(n)) * p0 / (1 - rho)**2
33
    # 计算平均等待时间在队列中 Wq
34
    wq = lq / lambda_
35
    # 计算系统中的总平均客户数 L
36
    l = lq + lambda_ / mu
37
    # 计算系统中的总平均等待时间 W
38
    w = wq + 1 / mu
39

40
    return l, w, lq, wq
41

42

43

44
def plot(x, y,  y_label):
45
    # 绘制图表
46
    plt.figure()
47
    # 绘制等待时间
48
    plt.plot(x, y[0], label="M/M/1", marker='o')
49
    plt.plot(x, y[1], label="M/M/n", marker='o')
50
    plt.title(y_label+" vs $\\rho$")
51
    plt.xlabel("$\\rho$")
52
    plt.ylabel(y_label)
53
    plt.legend()
54
    plt.grid(True)
55

56

57
# 利用率从0.1到0.9
58
rho_values = [0.1,0.2,0.3,0.4,0.5,0.6,0.7,0.8,0.85,0.9]
59
mu = 1
60
# 服务台数量
61
n = 5        
62

63
# 客户到达率
64
lambda_ = rho_values*mu  
65
mm1_array = np.zeros((len(rho_values), 4))
66
mmn_array = np.zeros((len(rho_values), 4))
67

68
for i in range(len(rho_values)):
69
    lam = lambda_[i]
70
    mm1_array[i] = mm1_queue(lam, mu)
71
    mmn_array[i] = mmn_queue(n*lam, mu, n)
72

73
plt.figure()
74
# 绘制队列长度
75
plt.plot(rho_values, n*mm1_array[:, 0], label="n$\\times$M/M/1 L", marker='o', color='r')
76
plt.plot(rho_values, mmn_array[:, 0], label="M/M/n L", marker='o', color='g')
77
plt.plot(rho_values, n*mm1_array[:, 2],
78
         label="n$\\times$M/M/1 Lq", marker='*', color='r')
79
plt.plot(rho_values, mmn_array[:, 2], label="M/M/n Lq", marker='*', color='g')
80
plt.title("$L,Lq$"+" vs $\\rho$")
81
plt.xlabel("$\\rho$")
82
plt.ylabel("$L,Lq$")
83
plt.legend()
84
plt.grid(True)
85

86
plt.figure()
87
# 绘制等待时间
88
plt.plot(rho_values, mm1_array[:, 1], label="n$\\times$M/M/1 W", marker='o', color='r')
89
plt.plot(rho_values, mmn_array[:, 1], label="M/M/n W", marker='o', color='g')
90
plt.plot(rho_values, mm1_array[:, 3],
91
         label="n$\\times$M/M/1 Wq", marker='*', color='r')
92
plt.plot(rho_values, mmn_array[:, 3], label="M/M/n Wq", marker='*', color='g')
93
plt.title("$W,Wq$"+" vs $\\rho$")
94
plt.xlabel("$\\rho$")
95
plt.ylabel("$W,Wq$")
96
plt.legend()
97
plt.grid(True)
98

99
plt.show()
100

参考