重平衡问题

标准的重平衡算法是 kafka 的重平衡算法。

可以覆写的重平衡算法

默认生产者行为

每个机房的生产者优先生产到本机房的 broker。

默认策略

同机房集群—>同地域跨机房集群—>跨地域集群

同机房集群优先

向同机房的全部集群发送消息。

某些 mq 的实现不能向全部集群发送消息，会按照一个负载均衡列表按顺序选择集群，控制这个负载均衡列表实际上就能控制负载均衡策略。

同地域集群优先

优先向同地域内的全部集群发送消息。

某些 mq 的实现不能向全部集群发送消息，会按照一个负载均衡列表按顺序选择集群，控制这个负载均衡列表实际上就能控制负载均衡策略。

全部集群

不区分服务端集群的机房信息，向全部集群发送消息。

默认消费者行为

所有消费者都可以参与所有集群的消费。

指定分配

不要轻易指定这个策略。

这个策略有它的危险性：指定 partition 消费，服务端将不对该消费组的 partition 进行分配，完全交给客户端负责，如果客户端死掉，可能出现 partition 无人消费的情况。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// 创建topic对应的consumer对象（注意每次build调用会产生一个新的实例）
       List<Integer> partitionList = new LinkedList<>();
       partitionList.add(1);
       partitionList.add(4);
       IConsumerProcessor consumer = MqClient.buildConsumerFactory(properties, "common_partition", partitionList);
       
// 调用recvMessageWithParallel设置listener
       // 注意1：可以修改String.class以支持自定义数据类型
       // 注意2：针对同一个consumer对象，只能调用一次该方法；多次调用的话，后面的调用都会报异常
       consumer.recvMessageWithParallel(String.class, new IMessageListener() {
           @Override
           public ConsumeStatus recvMessage(MqMessage message, MessagetContext context) {
               try {
                   System.out.println("message=[" + message.getBody() + "]  partition=" + message.getParttion());
               } catch (Exception e) {
                   e.printStackTrace();
               }
               return ConsumeStatus.CONSUME_SUCCESS;
           }
       });

       //如上简单demo可以达到消费者一直处于消费状态,业务可以主动调用consumer.close()来关闭consumer,如消费100个消息主动关闭

同机房优先/地域敏感

尽可能降低由于网络传输带来的时延问题：按照同机房优先、同地域优先、跨地域优先。从前到后优先级越来越低的方式进行参与者划分；具体分配方式仍使用原有的集群粒度 partition 分配。

每次只有一种策略被激活：

  • 同机房优先不会激活同地域优先：如果说本 broker 有同机房的消费者，哪怕这个消费者的消费能力比较差，负载均衡也不会让本 broker 的 partition 被跨机房的消费者消费，这会导致积压。如果激活同地域优先，则同机房的 broker 能够被同机房的 consumer 消费，同地域的 consumer 可以消费无主的 broker-某些“地域敏感”实现更进一步，即使 broker 可以被同机房消费，也可以被同地域消费。
  • 同地域优先不会激活跨地域优先：如果说本 broker 有同地域的消费者，哪怕这个消费者的消费能力比较差，负载均衡也不会让本 broker 的 partition 被跨地域的消费者消费，这会导致积压。

积压问题

可能出现积压的几种原因：

  1. 客户端没有正确启动。
  2. 客户端虽然启动了，但出现消息卡单-这种情况可以通过 JStack 发现问题。
  3. 消费能力不足：更下游有些慢阻塞操作，阻塞了 ConsumerFetcherThread ，特别是更下游如果有阻塞服务的话问题更大。

延迟问题

消费组级延迟

消费组延迟生效在消费端，大概原理可以理解为客户端收到消息之后延迟一定时间之后再调用业务消费逻辑进行消费。

这种方法的实现通常不用改代码，但可能需要消费端有个缓冲区。如果缓冲区的大小固定，则消息可能会丢失。

消息级延迟

消息粒度延迟的原理大概是客户端在发送消息之后，不会马上发送到broker,而是先把消息暂存到带有 zset 之类的数据结构的 kv 存储中，等延时时间到达才正式发送消息到broker。每条消息的延迟时间可以不同，通过代码控制。