RocketMQ - 常见题目

RocketMQ常见题目


  
一：消息幂等题目

1：什么是消费幂等

重复消费的结果与消费一次的结果是相同的，而且多次消费并未对业务系统产生任何负面影响，那么这个消费过程就是消费幂等的。
   幂等：若某操作执行多次与执行一次对系统产生的影响是相同的，则称该操作是幂等的。
  在互联网应用中，尤其在网络不稳固的环境下，消息很有大概会出现重复发送或重复消费。
如果重复的消息大概会影响业务处理，那么就应该对消息做幂等处理。
2：消息重复的场景分析

什么环境下大概会出现消息被重复消费呢？最常见的有以下三种环境：
2.1：发送时消息重复

当一条消息已被乐成发送到Broker并完成长期化，此时出现了网络闪断，从而导致Broker对Producer应答失败。
如果此时Producer意识到消息发送失败并实验再次发送消息，此时Broker中就大概会出现两条内容相同而且Message ID也相同的消息，那么后续Consumer就一定会消费两次该消息。
2.2：消费时消息重复

消息已投递到Consumer并完成业务处理，当Consumer给Broker反馈应答时网络闪断，Broker没有接收到消费乐成响应。
为了保证消息至少被消费一次的原则，Broker将在网络规复后再次实验投递之前已被处理过的消息。
此时消费者就会收到与之前处理过的内容相同、Message ID也相同的消息。
2.3：Rebalance时消息重复

当Consumer Group中的Consumer数量发生变革时，或其订阅的Topic的Queue数量发生变革时，会触发Rebalance
此时Consumer大概会收到曾经被消费过的消息。
3：通用办理方案

   重复消费没有办法剔除，但是要保证重复的消费幂等 <- 所以办理方案都是围绕保证幂等性的
  3.1：两要素(幂等令牌 & 唯一性处理)

幂等办理方案的计划中涉及到两项要素：幂等令牌 & 唯一性处理。只要充分使用好这两要素，就可以计划出好的幂等办理方案。

• 幂等令牌：是生产者和消费者两者中的既定协议，通常指具备唯一业务标识的字符串。
  
  
  
  • 例如，订单号、流水号。一般由Producer随着消息一同发送来的。
    
  
  
• 唯一性处理：服务端通过接纳一定的算法策略，保证同一个业务逻辑不会被重复执行乐成多次。
  
  
  
  • 例如，对同一笔订单的多次付出操作，只会乐成一次。
    
  
  

3.2：通用办理方案

对于常见的系统，幂等性操作的通用性办理方案是：

• 起首通过缓存去重。在缓存中如果已经存在了某幂等令牌，则说明本次操作是重复性操作；若缓存没有掷中，则进入下一步。
  
• 在唯一性处理之前，先在数据库中查询幂等令牌作为索引的数据是否存在。若存在，则说明本次操作为重复性操作；若不存在，则进入下一步。
  
• 在同一事件中完成三项操作：唯一性处理后，将幂等令牌写入到缓存，并将幂等令牌作为唯一索引的数据写入到DB中。
  

   第 1 步已经判定过是否是重复性操作了，为什么第 2 步还要再次判定？
  一般缓存中的数据是具有有用期的。缓存中数据的有用期一旦逾期，就是发生缓存穿透，使哀求直接就到达了DBMS。
  3.3：办理方案举例

以付出场景为例：

• 当付出哀求到达后，起首在Redis缓存中却获取key为付出流水号的缓存value。
  
  
  
  • 若value不空，则说明本次付出是重复操作，业务系统直接返回调用侧重复付出标识；
    
  • 若value为空，则进入下一步操作
    
  
  
• 到DBMS中根据付出流水号查询是否存在相应实例。
  
  
  
  • 若存在，则说明本次付出是重复操作，业务系统直接返回调用侧重复付出标识；
    
  • 若不存在，则说明本次操作是初次操作，进入下一步完成唯一性处理
    
  
  
• 在分布式事件中完成三项操作：
  
  
  
  • 完成付出使命
    
  • 将当前付出流水号作为key，任意字符串作为value，通过set(key, value, expireTime)将数据写入到Redis缓存
    
  • 将当前付出流水号作为主键，与别的相干数据共同写入到DBMS
    
  
  

3.4：消息幂等的实现

消费幂等的办理方案很简单：为消息指定不会重复的唯一标识。因为Message ID有大概出现重复的环境，所以真正安全的幂等处理，不发起以Message ID作为处理依据。
最好的方式是以业务唯一标识作为幂等处理的关键依据，而业务的唯一标识可以通过消息Key设置。
以付出场景为例，可以将消息的Key设置为订单号，作为幂等处理的依据。具体代码示例如下：

1. Message message = new Message();
2. message.setKey("ORDERID_100");
3. SendResult sendResult = producer.send(message);

复制代码

消费者收到消息时可以根据消息的Key即订单号来实现消费幂等：

1. consumer.registerMessageListener(new MessageListenerConcurrently() {
2.     @Override
3.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context ) {
4.         for(MessageExt msg : msgs) {
5.             String key = msg.getKeys();
6.             // 根据业务唯一标识Key做幂等处理
7.             // ......
8.             }
9.         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
10.     }
11. });

复制代码

RocketMQ能够保证消息不丢失，但不能保证消息不重复
二：消息堆积题目

消息处理流程中，如果Consumer的消费速度跟不上Producer的发送速度，MQ中未处理的消息会越来越多，这部门消息就被称为堆积消息。
消息出现堆积进而会造成消息的消费延长。以了局景须要重点关注消息堆积和消费延长题目：

• 业务系统上下游能力不匹配造成的持续堆积，且无法自行规复。
  
• 业务系统对消息的消费及时性要求较高，即使是短暂的堆积造成的消费延长也无法接受。
  

1：产生原因分析

Consumer使用长轮询Pull模式消费消息时，分为以下两个阶段：
1.1：消息拉取

consumer通过长轮询Pull模式批量拉取的方式从服务端获取消息，将拉取的消息缓存到当地缓冲队列中
对于拉取式消费，在内网环境下会有很高的吞吐量，所以这一阶段一般不会是消息堆积的瓶颈
   一个单线程分区的低规格主机其可达到几万的TPS，如果是多个分区的多个线程，可以轻松达到几十万的TPS
  1.2：消息消费

Consumer将当地缓存的消息提交到消费线程中，使用业务消费逻辑对消息进行处理，处理完毕后获取到一个结果。这是真正的消息消费过程。
此时Consumer的消费能力就完全依赖于消息的消费耗时和消费并发度了。
如果由于业务处理逻辑复杂等原因，导致处理单条消息的耗时较长，则团体的消息吞吐量肯定不会高，此时就会导致Consumer当地缓冲队列达到上限，制止从服务端拉取消息。
所以，消息堆积的重要瓶颈在于客户端的消费能力，而消费能力由消费耗时和消费并发度决定。
留意，消费耗时的优先级要高于消费并发度。即在保证了消费耗时的合理性前提下，再考虑消费并发度题目。
2：消费耗时

影响消息处理时长的重要因素是代码逻辑。而代码逻辑中大概会影响处理时长代码重要有两种范例：CPU内部计算型代码和外部I/O操作型代码。
通常环境下代码中如果没有复杂的递归和循环的话，内部计算耗时相对外部I/O操作来说几乎可以忽略。所以外部IO型代码是影响消息处理时长的重要症结地点。
   外部IO操作型代码举例：
  

• 读写外部数据库，例如对远程MySQL的访问
  
• 读写外部缓存系统，例如对远程Redis的访问
  
• 下游系统调用，例如Dubbo的RPC远程调用，Spring Cloud的对下游系统的Http接口调用
  

  关于下游系统调用逻辑须要进行提前梳理，把握每个调用操作预期的耗时，这样做是为了能够判定消费逻辑中IO操作的耗时是否合理。
  通常消息堆积是由于下游系统出现了服务异常或达到了DBMS容量限制，导致消费耗时增加。
  服务异常，并不仅仅是系统中出现的类似 500 这样的代码错误，而大概是更加潜伏的题目。例如，网络带宽题目。
  达到了DBMS容量限制，其也会引发消息的消费耗时增加。
  3：消费并发度

一般环境下，消费者端的消费并发度由单节点线程数和节点数量共同决定，其值为value = 单节点线程数*节点数量。
通常须要优先调解单节点的线程数，若单机硬件资源达到了上限，则须要通过横向扩展来进步消费并发度。
   单节点线程数，即单个Consumer所包含的线程数量
  节点数量，即Consumer Group所包含的Consumer数量
  对于普通消息、延时消息及事件消息，并发度计算都是单节点线程数*节点数量。但对于顺序消息则是差别的。
  顺序消息的消费并发度等于Topic的Queue分区数量。
全局顺序消息：

• 该范例消息的Topic只有一个Queue分区。其可以保证该Topic的所有消息被顺序消费。
  
• 为了保证这个全局顺序性，Consumer Group中在同一时刻只能有一个Consumer的一个线程进行消费。所以其并发度为 1 。
  

分区顺序消息：

• 该范例消息的Topic有多个Queue分区。其仅可以保证该Topic的每个Queue分区中的消息被顺序消费，不能保证整个Topic中消息的顺序消费。
  
• 为了保证这个分区顺序性，每个Queue分区中的消息在Consumer Group中的同一时刻只能有一个Consumer的一个线程进行消费。
  
• 在同一时刻最多会出现多个Queue分蘖有多个Consumer的多个线程并行消费。所以其并发度为Topic的分区数量。
  

4：单机线程的计算

对于一台主机中线程池中线程数的设置须要谨慎，不能盲目直接调大线程数，设置过大的线程数反而会带来大量的线程切换的开销。
抱负环境下单节点的最优线程数计算模型为：C *（T1 + T2）/ T1。

• C：CPU内核数
  
• T1：CPU内部逻辑计算耗时
  
• T2：外部IO操作耗时
  

   最优线程数 = C *（T1 + T2）/ T1 = C * T1/T1 + C * T2/T1 = C + C * T2/T1
    留意，该计算出的数值是抱负状态下的理论数据，在生产环境中，不发起直接使用。而是根据当前环境，先设置一个比该值小的数值然后观察其压测结果，然后再根据结果渐渐调大线程数，直至找到在该环境中性能最佳时的值。
  5：怎样避免消息堆积和消费延长

为了避免在业务使用时出现非预期的消息堆积和消费延长题目，须要在前期计划阶段对整个业务逻辑进行完善的排查和梳理。
此中最重要的就是梳理消息的消费耗时和设置消息消费的并发度。
5.1：梳理消息的消费耗时

通过压测获取消息的消费耗时，并对耗时较高的操作的代码逻辑进行分析。梳理消息的消费耗时须要关注以下信息：

• 消息消费逻辑的计算复杂度是否过高，代码是否存在无限循环和递归等缺陷。
  
• 消息消费逻辑中的I/O操作是否是必须的，可否用当地缓存等方案规避。
  
• 消费逻辑中的复杂耗时的操作是否可以做异步化处理。如果可以，是否会造成逻辑错乱。
  

5.2：设置消费并发度

对于消息消费并发度的计算，可以通过以下两步实施：

• 渐渐调大单个Consumer节点的线程数，并观测节点的系统指标，得到单个节点最优的消费线程数和消息吞吐量。
  
• 根据上下游链路的流量峰值计算出须要设置的节点数
  

   节点数 = 流量峰值 / 单个节点消息吞吐量
  三：消息清理题目

消息被消费过后会被清理掉吗？不会的。
消息是被顺序存储在commitlog文件的，且消息巨细不定长，所以消息的清理是不大概以消息为单元进行清理的
消息是以commitlog文件为单元进行清理的。否则会急剧降落清理效率，并实现逻辑复杂。
commitlog文件存在一个逾期时间，默认 72 小时
除了用户手动清理外，在以下环境下也会被主动清理，无论文件中的消息是否被消费过：

• 文件逾期，且到达清理时间点（默认为破晓 4 点）后，主动清理逾期文件
  
• 文件逾期，且磁盘空间占用率已达逾期清理警戒线（默认75%）后，无论是否达到清理时间点，都会主动清理逾期文件
  
• 磁盘占用率达到清理警戒线（默认85%）后，开始按照设定好的规则清理文件，无论是否逾期。默认会从最老的文件开始清理
  
• 磁盘占用率达到系统伤害警戒线（默认90%）后，Broker将拒绝消息写入
  

须要留意以下几点：

• 对于RocketMQ系统来说，删除一个1G巨细的文件，是一个压力巨大的IO操作。
  
  
  
  • 在删除过程中，系统性能会骤然降落。所以，其默认清理时间点为破晓 4 点，访问量最小的时间。
    
  • 也正因如果，我们要保障磁盘空间的空闲率，不要使系统出如今别的时间点删除commitlog文件的环境。
    
  
  
• 官方发起RocketMQ服务的Linux文件系统接纳ext4。因为对于文件删除操作，ext4要比ext3性能更好
  

四：消息过滤题目

消息者在进行消息订阅时，除了可以指定要订阅消息的Topic外，还可以对指定Topic中的消息根据指定条件进行过滤，即可以订阅比Topic更加细粒度的消息范例。
对于指定Topic消息的过滤有两种过滤方式：Tag过滤与SQL过滤。
1：Tag方式过滤-简单

通过consumer的subscribe()方法指定要订阅消息的Tag。如果订阅多个Tag的消息，Tag间使用或运算符(双竖线||)毗连。

1. DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
2. consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

复制代码

2：SQL过滤-功能强盛

SQL过滤是一种通过特定表达式对事先埋入到消息中的用户属性进行筛选过滤的方式。
通过SQL过滤，可以实现对消息的复杂过滤。不过，只有使用PUSH模式的消费者才能使用SQL过滤。
SQL过滤表达式中支持多种常量范例与运算符。
支持的常量范例：

• 数值：比如： 123 ，3.1415
  
• 字符：必须用单引号包裹起来，比如：‘abc’
  
• 布尔：TRUE 或 FALSE
  
• NULL：特别的常量，表示空
  

支持的运算符有：

• 数值比力：>，>=，<，<=，BETWEEN，=
  
• 字符比力：=，<>，IN
  
• 逻辑运算 ：AND，OR，NOT
  
• NULL判定：IS NULL 或者 IS NOT NULL
  

默认环境下Broker没有开启消息的SQL过滤功能，须要在Broker加载的配置文件中添加如部属性，以开启该功能：

1. enablePropertyFilter = true

复制代码

在启动Broker时须要指定这个修改过的配置文件。
例如对于单机Broker的启动，其修改的配置文件是conf/broker.conf，启动时使用如下命令：

1. sh bin/mqbroker -n localhost:9876 -c conf/broker.conf &

复制代码

3：代码举例

3.1：Tag方式进行过滤例子

   界说Tag过滤Producer

1. public class FilterByTagProducer {
2.     public static void main(String[] args) throws Exception {
3.         DefaultMQProducer producer = new DefaultMQProducer("pg");
4.         producer.setNamesrvAddr("rocketmqOS:9876");
5.         producer.start();
6.         String[] tags = {"myTagA","myTagB","myTagC"};
7.         for (int i = 0 ; i < 10 ; i++) {
8.             byte[] body = ("Hi," + i).getBytes();
9.             String tag = tags[i % tags.length];
10.             // 指定tag
11.             Message msg = new Message("myTopic", tag, body);
12.             SendResult sendResult = producer.send(msg);
13.             System.out.println(sendResult);
14.         }
15.         producer.shutdown();
16.     }
17. }

复制代码

  界说Tag过滤Consumer

1. public class FilterByTagConsumer {
2.     public static void main(String[] args) throws Exception {
3.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("pg");
4.         consumer.setNamesrvAddr("rocketmqOS:9876"); // 设置NameServer
5.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
6.         // 消费者订阅指定的tag, 多个tag用||分割
7.         consumer.subscribe("myTopic", "myTagA || myTagB"); // 订阅主题和tag，只有tagA或者tagB才进行消费
8.         // 定义异步并发监听，准备消费
9.         consumer.registerMessageListener(new MessageListenerConcurrently() {
10.             @Override
11.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {
12.                 for (MessageExt me:msgs){
13.                     System.out.println(me);
14.                 }
15.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
16.             }
17.         });
18.         consumer.start();
19.         System.out.println("Consumer Started");
20.     }
21. }

复制代码

3.2：SQL方式进行过滤例子

   界说SQL过滤Producer

1. public class FilterBySQLProducer {
2.     public static void main(String[] args) throws Exception {
3.         DefaultMQProducer producer = new DefaultMQProducer("pg");
4.         producer.setNamesrvAddr("rocketmqOS:9876");
5.         producer.start();
6.         for (int i = 0 ; i < 10 ; i++) {
7.             try {
8.                 byte[] body = ("Hi," + i).getBytes();
9.                 Message msg = new Message("myTopic", "myTag", body);
10.                 // 自己声明一个属性，并赋值，到时候sq
11.                 msg.putUserProperty("age", i + "");
12.                 SendResult sendResult = producer.send(msg);
13.                 System.out.println(sendResult);
14.             } catch (Exception e) {
15.                 e.printStackTrace();
16.             }
17.         }
18.         producer.shutdown();
19.     }
20. }

复制代码

  界说SQL过滤Consumer

1. public class FilterBySQLConsumer {
2.     public static void main(String[] args) throws Exception {
3.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("pg");
4.         consumer.setNamesrvAddr("rocketmqOS:9876");
5.         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
6.         // 注意下面这句
7.         consumer.subscribe("myTopic", MessageSelector.bySql("age between 0 and 6"));
8.         
9.         // 监听并
10.         consumer.registerMessageListener(new MessageListenerConcurrently() {
11.             @Override
12.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
13.                 for (MessageExt me:msgs){
14.                     System.out.println(me);
15.                 }
16.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
17.             }
18.         });
19.         consumer.start();
20.         System.out.println("Consumer Started");
21.     }
22. }

复制代码

五：消息重试题目

1：消息发送的重试机制

Producer对发送失败的消息进行重新发送的机制，称为消息发送重试机制，也称为消息重投机制。
   留意事项
  

• 生产者在发送消息时，若接纳同步或异步发送方式，发送失败会重试，但oneway消息发送方式发送失败是没有重试机制的
  
• 只有普通消息具有发送重试机制，顺序消息是没有的
  
• 消息重投机制可以保证消息尽大概发送乐成、不丢失，但大概会造成消息重复。消息重复在RocketMQ中是无法避免的题目
  
• 消息发送重试有三种策略可以选择：同步发送失败策略、异步发送失败策略、消息刷盘失败策略
  

1.1：同步发送失败策略

对于普通消息，消息发送默认接纳round-robin策略来选择所发送到的队列。
如果发送失败，默认重试 2 次。

• 重试时是不会选择上次发送失败的Broker，而是选择别的Broker。
  
• 若只有一个Broker其也只能发送到该Broker，但其会尽量发送到该Broker上的别的Queue。
  

1. // 创建一个producer，参数为Producer Group名称
2. DefaultMQProducer producer = new DefaultMQProducer("pg");
3. // 指定nameServer地址
4. producer.setNamesrvAddr("rocketmqOS:9876");
5. // 设置同步发送失败时重试发送的次数，默认为 2 次
6. producer.setRetryTimesWhenSendFailed( 3 );
7. // 设置发送超时时限为5s，默认3s
8. producer.setSendMsgTimeout( 5000 );

复制代码

同时，Broker还具有失败隔离功能，使Producer尽量选择未发生过发送失败的Broker作为目标Broker。
这样做可以保证别的消息尽量不发送到题目Broker，为了提升消息发送效率，降低消息发送耗时。
   思考：让我们自己实现失败隔离功能，怎样来做？
  

• 方案一：Producer中维护某JUC的Map聚集，其key是发生失败的时间戳，value为Broker实例。Producer中还维护着一个Set聚集，此中存放着所有未发生发送异常的Broker实例。选择目标Broker是从该Set聚集中选择的。再界说一个定时使命，定期从Map聚集中将恒久未发生发送异常的Broker清理出去，并添加到Set聚集。
  
• 方案二：为Producer中的Broker实例添加一个标识，例如是一个AtomicBoolean属性。只要该Broker上发生过发送异常，就将其置为true。选择目标Broker就是选择该属性值为false的Broker。再界说一个定时使命，定期将Broker的该属性置为false。
  
• 方案三：为Producer中的Broker实例添加一个标识，例如是一个AtomicLong属性。只要该Broker上发生过发送异常，就使其值增一。选择目标Broker就是选择该属性值最小的Broker。若该值相同，接纳轮询方式选择。
  

  如果超过重试次数，则抛出异常，由Producer去保证消息不丢。
当然当生产者出现RemotingException、MQClientException和MQBrokerException时，Producer会主动重投消息。
1.2：异步发送失败策略

异步发送失败重试时，异步重试不会选择其他broker，仅在同一个broker上做重试，所以该策略无法保证消息不丢。

1. DefaultMQProducer producer = new DefaultMQProducer("pg");
2. producer.setNamesrvAddr("rocketmqOS:9876");
3. // 指定异步发送失败后不进行重试发送
4. producer.setRetryTimesWhenSendAsyncFailed( 0 );

复制代码

1.3：消息刷盘失败策略

消息刷盘超时或slave不可用（slave在做数据同步时向master返回状态不是SEND_OK）时，默认是不会将消息实验发送到其他Broker的。
对于重要消息可以通过在Broker的配置文件设置retryAnotherBrokerWhenNotStoreOK = true来开启。
2：消费重试

2.1：顺序消息的消费重试

对于顺序消息，当Consumer消费消息失败后，为了保证消息的顺序性，其会主动不断地进行消息重试，直到消费乐成。
消费重试默认隔断时间为 1000 毫秒。重试期间应用会出现消息消费被阻塞的环境。

1. DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
2. // 顺序消息消费失败的消费重试时间间隔，单位毫秒，默认为 1000 ，其取值范围为[10,30000]
3. consumer.setSuspendCurrentQueueTimeMillis( 100 );

复制代码

  由于对顺序消息的重试是无休止的，不停止的，直至消费乐成
  ⚠️ 所以，对于顺序消息的消费，务须要保证应用能够及时监控并处理消费失败的环境，避免消费被永久性阻塞。
  ⚠️ 顺序消息没有发送失败重试机制，但具有消费失败重试机制
  2.2：无序消息的消费重试

对于无序消息（普通消息、延时消息、事件消息），当Consumer消费消息失败时，可以通过设置返回状态达到消息重试的结果。
不过须要留意，无序消息的重试只对集群消费方式生效，广播消费方式不提供失败重试特性。
即对于广播消费，消费失败后，失败消息不再重试，继续消费后续消息。
2.3：消息重试次数和隔断

对于无序消息集群消费下的重试消费，每条消息默认最多重试 16 次，但每次重试的隔断时间是差别的，会渐渐变长。
每次重试的隔断时间如下表。
重试次数与上次重试的隔断时间重试次数与上次重试的隔断时间110秒97分钟230108 分钟31分钟119 分钟42分钟1210分钟53分钟1320分钟64分钟1430分钟75分钟151小时86分钟162 小时   若一条消息在一直消费失败的前提下，将会在正常消费后的第 4 小时 46 分后进行第 16 次重试。
  若仍旧失败，则将消息投递到死信队列
  修改消费重试次数

1. DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
2. // 修改消费重试次数
3. consumer.setMaxReconsumeTimes( 10 );

复制代码

  对于修改过的重试次数，将按照以下策略执行：
  

• 若修改值 < 16 ，则按照指定隔断进行重试
  
• 若修改值 > 16 ，则超过 16 次的重试时间隔断均为 2 小时
  

  对于Consumer Group
  

• 若仅修改了一个Consumer的消费重试次数，则会应用到该Group中所有别的Consumer实例。
  
• 若出现多个Consumer均做了修改的环境，则接纳覆盖方式生效。即末了被修改的值会覆盖前面设置的值。
  

  2.4：重试队列

对于须要重试消费的消息，并不是Consumer在等候了指定时长后再次去拉取原来的消息进行消费，而是将这些须要重试消费的消息放入到了一个特别Topic的队列中，而后进行再次消费的。这个特别的队列就是重试队列。
当出现须要进行重试消费的消息时，Broker会为每个消费组都设置一个Topic名称为%RETRY%consumerGroup@consumerGroup的重试队列。
   

• 这个重试队列是针对消息才组的，而不是针对每个Topic设置的
  
• 只有当出现须要进行重试消费的消息时，才会为该消费者组创建重试队列
  

  

   留意，消费重试的时间隔断与延时消费的延时品级十分相似
  除了没有延时品级的前两个时间外，别的的时间都是相同的
  Broker对于重试消息的处理是通过延时消息实现的。

• 先将消息保存到SCHEDULE_TOPIC_XXXX延长队列中
  
• 延长时间到后，会将消息投递到%RETRY%consumerGroup@consumerGroup重试队列中。
  

2.5：消息重试/不重试配置方式

2.5.1：重试配置

集群消费方式下，消息消费失败后若希望消费重试，则须要在消息监听器接口的实现中明确进行如下三种方式之一的配置：

• 返回ConsumeConcurrentlyStatus.RECONSUME_LATER（保举）
  
• 返回Null
  
• 抛出异常
  

1. @Override
2. public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
3.     try {
4.         ...
5.     } catch (Throwable e) {
6.         // 以下三种方式都可以引发消息重试
7.         return ConsumeConcurrentlyStatus.RECONSUME_LATER;
8.         // return null;
9.         // throw new RuntimeException("消费异常");
10.     }
11.    
12.     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
13. }

复制代码

2.5.2：不重试配置

在捕捉到异常后同样也返回与消费乐成后的相同的结果，即ConsumeConcurrentlyStatus.CONSUME_SUCCESS，则不进行消费重试。

1. @Override
2. public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
3.     try {
4.         ...
5.     } catch (Throwable e) {
6.         // 也return成功，就不进行重试
7.         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
8.     }
9.    
10.     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
11. }

复制代码

2.6：死信队列

当一条消息初次消费失败，消息队列会主动进行消费重试；达到最大重试次数后，若消费依然失败，则表明消费者在正常环境下无法精确地消费该消息，此时，消息队列不会立刻将消息丢弃，而是将其发送到该消费者对应的特别队列中。这个队列就是死信队列（Dead-Letter Queue，DLQ）
死信队列此中的消息称为死信消息（Dead-Letter Message，DLM）。
   死信队列是用于处理无法被正常消费的消息的。
  2.6.1：死信队列特性

• 死信队列中的消息不会再被消费者正常消费，即DLQ对于消费者是不可见的
  
• 死信存储有用期与正常消息相同，均为 3 天（commitlog文件的逾期时间）， 3 天后会被主动删除
  
• 死信队列就是一个特别的Topic，名称为%DLQ%consumerGroup@consumerGroup，即每个消费者组都有一个死信队列
  
• 如果一个消费者组未产生死信消息，则不会为其创建相应的死信队列
  
• 一个死信队列包含了对应Group ID产生的所有死信消息，岂论该消息属于哪个Topic。
  

2.6.2：查询死信消息

消息队列RocketMQ版提供的查询死信消息的方式对比如下表所示。
查询方式查询条件查询类别说明按Group ID查询Group ID + 时间段范围查询根据Group ID和时间范围，批量获取符合条件的所有消息；
查询量大，不易匹配。按Message ID查询Group ID + Message ID准确查询根据Group ID和Message ID可以准确定位任意一条消息。

• 登录消息队列RocketMQ版控制台，在左侧导航栏单击实例列表。
  
• 在顶部菜单栏选择地区，如华东1（杭州），然后在实例列表中，单击目标实例名称。
  
• 在左侧导航栏单击死信队列，然后在死信队列页面，选择以下任一方式查询死信消息：
  

• 按 Group 查询 -> 根据Group ID和死信消息产生的时间范围，批量查询该Group ID在某段时间内产生的所有死信消息。页面中会显示所有符合筛选条件的死信消息
  

• 按Message ID查询消息属于准确查询。您可以根据Group ID与Message ID准确查询到任意一条消息。页面中会显示所有符合筛选条件的死信消息
  

2.6.3：死信队列处理

实际上，当一条消息进入死信队列，就意味着系统中某些地方出现了题目，从而导致消费者无法正常消费该消息，比如代码中本来就存在Bug。
因此须要开辟职员进行特别处理。最关键的步骤是要排查可疑因素，办理代码中大概存在的Bug，然后再将原来的死信消息再次进行投递消费
六：怎样保证消息不丢失

1：Producer怎样保证消息不丢失

​
生产者发送消息之后，给生产者一个确定的通知，这个消息在Broker端是否写入完成了。

1. //异步发送，不需要broker确认，效率高，但是会有丢消息的可能。
2. producer.sendOneway(msg);
3. //同步发送，生产者等待Broker确认。消息最安全，但效率最低
4. SendResult sendResult = producer.send(msg,20*1000);
5. //异步发送，生产者另起一个线程等待broker确认，收到Broker确认之后直接触发回调方法。消息安全和效率之间比较均衡，但是会加大客户端的负担。
6. producer.send(msg,new SendCallback(){
7.         @Override
8.         public void onSuccess(SendResult sendResult){
9.                 //do something
10.         }
11.        
12.         @Override
13.         public void onException(Throwable e){
14.                 //do something
15.         }
16. })

复制代码

​与之类似的，Kafka同样也提供了这种同步和异步的发送消息机制

1. //直接send发送消息，返回的是一个Future。这就相当于是异步调用
2. Future<RecordMetadata> future = producer.send(record);
3. //调用future的get方法才会世纪获取到发送的结果，生产者收到这个结果后，就可以知道消息是否成功发送到broker了。这个过程就变成了一个同步的过程
4. RecordMetadata recordMetadata = producer.send(record).get();

复制代码

​而在RabbitMQ中，则是提供了一个Publisher Confirms生产者确认机制。
其思绪也是Publisher收到Broker的响应后再发出对应的回调方法。

1. //获取channel
2. Channel ch = ...;
3. //添加两个回调，一个处理ack响应，一个处理nack响应
4. ch.addConfirmListener(ConfirmCallback ackCallback,ConfirmCallback nackCallback)

复制代码

2：事件消息保证消息不丢失

先界说当地事件监听器

1. package com.mytest.mqdemo.producer;
3. import org.apache.commons.lang3.StringUtils;
4. import org.apache.rocketmq.client.producer.LocalTransactionState;
5. import org.apache.rocketmq.common.message.Message;
6. import org.apache.rocketmq.common.message.MessageExt;
8. /**
9. * <p>
10. * 功能描述：
11. * </p>
12. *
13. * @author cui haida
14. * @date 2024/03/24/17:52
15. */
16. public class TransactionListener implements org.apache.rocketmq.client.producer.TransactionListener {
17.     /**
18.      * 执行本地事务
19.      * @param message 消息体
20.      * @param o 参数
21.      * @return 执行本地事务的结果
22.      */
23.     @Override
24.     public LocalTransactionState executeLocalTransaction(Message message, Object o) {
25.         // todo: 执行本地事务，这里这是模拟
26.         String tags = message.getTags();
27.         if (StringUtils.contains("A", tags)) {
28.             return LocalTransactionState.COMMIT_MESSAGE;
29.         }
30.         if (StringUtils.contains("B", tags)) {
31.             return LocalTransactionState.ROLLBACK_MESSAGE;
32.         } else {
33.             return LocalTransactionState.UNKNOW;
34.         }
35.     }
37.     /**
38.      * 回查本地事务
39.      * @param message 待回查的本地事务信息
40.      * @return 回查结果
41.      */
42.     @Override
43.     public LocalTransactionState checkLocalTransaction(MessageExt message) {
44.         String tags = message.getTags();
45.         if (StringUtils.contains("C", tags)) {
46.             return LocalTransactionState.COMMIT_MESSAGE;
47.         }
48.         if (StringUtils.contains("D", tags)) {
49.             return LocalTransactionState.ROLLBACK_MESSAGE;
50.         } else {
51.             return LocalTransactionState.UNKNOW;
52.         }
53.     }
54. }

复制代码

生产者指定当地事件监听器，并进行消息的发送

1. package com.mytest.mqdemo.producer;
3. import org.apache.rocketmq.client.producer.TransactionMQProducer;
4. import org.apache.rocketmq.client.producer.TransactionSendResult;
5. import org.apache.rocketmq.common.message.Message;
7. import java.nio.charset.StandardCharsets;
8. import java.util.concurrent.ArrayBlockingQueue;
9. import java.util.concurrent.ThreadFactory;
10. import java.util.concurrent.ThreadPoolExecutor;
11. import java.util.concurrent.TimeUnit;
13. /**
14. * <p>
15. * 功能描述：
16. * </p>
17. *
18. * @author cui haida
19. * @date 2024/03/24/17:47
20. */
21. public class TransactionProducer {
22.     public static void main(String[] args) throws Exception {
23.         // 注意这里使用的是事务消息生产者
24.         TransactionMQProducer transactionProducer = new TransactionMQProducer("transaction_producer");
25.         transactionProducer.setNamesrvAddr("192.168.111.128:9876");
26.         // 异步提交，提高性能
27.         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
28.                 2, 5, 100, TimeUnit.SECONDS, // 4大基本参数
29.                 new ArrayBlockingQueue<>(2000), // 队列类型和大小
30.                 new ThreadFactory() { // 线程工厂指定
31.                     @Override
32.                     public Thread newThread(Runnable r) {
33.                         Thread thread = new Thread(r);
34.                         thread.setName("test-for-trans");
35.                         return thread;
36.                     }
37.                 });
38.         transactionProducer.setExecutorService(threadPoolExecutor);
39.         // 创建一个本地事务的监听器
40.         transactionProducer.setTransactionListener(new TransactionListener());
41.         transactionProducer.start();
42.         String[] tags = new String[]{"A", "B", "C", "D", "E"};
43.         for (int i = 0; i < 10; i++) {
44.             String body = (tags[i % tags.length] + "trans_message");
45.             Message message = new Message("trans", tags[i % tags.length], body.getBytes(StandardCharsets.UTF_8));
46.             // 发送
47.             TransactionSendResult transactionSendResult = transactionProducer.sendMessageInTransaction(message, null);
48.             System.out.println("消息发送成功_" + transactionSendResult);
49.             Thread.sleep(10);
50.         }
51.         Thread.sleep(10000);
52.         transactionProducer.shutdown();
53.     }
54. }

复制代码

界说消费者信息
因为半消息和回查对于消费者无感知，所以消费者就是最基本的实现

1. package com.mytest.mqdemo.customer;
3. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
4. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
5. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
6. import org.apache.rocketmq.client.exception.MQClientException;
7. import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
8. import org.apache.rocketmq.common.message.MessageExt;
10. import java.nio.charset.StandardCharsets;
12. /**
13. * @author cuihaida
14. * <p>
15. *     简单消费
16. * </p>
17. */
18. public class PushConsumer {
19.     public static void main(String[] args) throws MQClientException {
21.         // 消费的推拉模式
22.         // 拉模式：pull -> 消费者主动去Broker上拉取消息
23.         // 推模式：push -> 消费者等待Broker将消息推送过来
25.         // 定义一个push消费者
26.         String consumerGroup = "cg";
27.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
28.         // 指定nameServer
29.         consumer.setNamesrvAddr("192.168.111.129:9876");
30.         // 指定消费topic与tag
31.         String consumerTopic = "someTopic";
32.         String subTag = "*"; // 当前主题下的任意标签消息
33.         consumer.subscribe(consumerTopic, subTag);
35.         // 指定采用“广播模式”进行消费，默认为“集群模式”
36.         // consumer.setMessageModel(MessageModel.BROADCASTING);
38.         // 注册消息监听器
39.         // 一旦broker中有了其订阅的消息就会触发该方法的执行，其返回值为当前consumer消费的状态
41.         // MessageListener有两个实现类：
42.         // --> MessageListenerConcurrently -> 并发消费
43.         // --> MessageListenerOrderly -> 顺序消费
45.         // 第一个参数是消息列表，第二个参数是上下文
46.         consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
47.             // 逐条消费消息
48.             for (MessageExt msg : messages) {
49.                 try {
50.                     String msgBody = new String(msg.getBody(), StandardCharsets.UTF_8);
51.                     System.out.println(msgBody);
52.                     // todo: processMessage(msgBody);
53.                     // 手动ACK，返回消费状态：消费成功
54.                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
55.                 } catch (Exception e) {
56.                     System.out.println("消息消费异常");
57.                     // todo: 打印堆栈信息，异常处理
58.                     // 手动ACK，返回消费状态：需要重新消费
59.                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;
60.                 }
62.             }
63.             // 返回消费状态：消费成功
64.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
65.         });
67.         // 开启消费者消费
68.         consumer.start();
69.         System.out.println("=========> Consumer Started <=============");
70.     }
71. }

复制代码

3：broker保证消息不丢失

Producer把消息发送到Broker上了之后，Broker是不是能保证消息不丢失呢？这里有一个焦点题目——PageCache缓存
​ 数据会优先写入缓存，然后过一段时间再写图到磁盘。但是缓存中的数据有个特点，就是断点即丢失，所以，如果服务器发生非正常断电，内存中的数据还没有写入磁盘，这是就会造成消息丢失。
以Linux为例，用户态就是应用程序，不管是什么应用程序，想要写入磁盘文件时，都只能调用操作系统提供的write系统调用，申请写磁盘。至于消息怎样颠末PageCache再写入到磁盘中，这个过程就是在内核态执行的，也就是操作系统自己执行的，应用程序无法干预。这个过程中，应用系统唯一能够干预的，就是调用操作系统提供的sync系统调用，申请一次刷盘操作，主动将PageCache中的数据写入到磁盘。
4：Broker主从同步怎样保证不丢失

对于Broker来说，通常Slaver的作用就是做一个数据备份，当Broker服务宕机时，甚至是磁盘坏了时，可以从Slaver上获取数据记载。
但是，如果主从同步失败了，那么Broker的这一层保证就会失效。因此，主从同步也有大概造成消息的丢失。

• 普通集群 - 指定角色，各司其职
  
• Dledger高可用集群，自行选举，多数同意
  

5：Consumer怎样保证消费不丢失

几乎所有的MQ产品都设置了消费者消费者状态确认机制。也就是消费者处理完消息之后，须要给Broker一个响应，表示消息被正常处理了。
如果Broker没有拿到这个响应，不管是因为Consumer没有拿到还是Consumer处理完消息后咩有给出响应，Broker都会认为消息没有处理乐成。之后，Broker就会向Consumer重复投递这些没有处理乐成的消息。
6：如果MQ服务全挂了，怎样保证不丢失

​ 针对这种环境，通常做法是计划一个降级缓存。Producer往MQ发消息失败了，就往降级缓存中写，然后，依然正常去进行后续的业务。
​ 此时，再启动一个线程，不断实验将降级缓存中的数据往MQ中发送。这样，至少当MQ服务规复之后，这些消息可以尽快进入到MQ中，继续往下游Consumer推送，而不至于造成消息丢失。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。