RocketMQ顺序消息机制源码分析~

Myth丶恋晨 2023-10-15 09:41 89阅读 0赞

# 前言 #

本文来带大家了解一下RocketMQ的**顺序消息** ，虽然这玩意在实际开发中的**使用频率**不如普通消息，但是在某些场景下还必须用它~

例如前些日子的618，相信大家下了不少订单，银行卡疯狂的给你发余额短信，此时我们就要**注意了**！

举例: 当前你的银行卡余额为**100元**

1.  你先下了个单，购物金额为**20元**，此时银行卡余额为**80元**
2.  然后你又下了个单，购物金额为**50元**，此时银行卡余额为**30元**

这时候就要注意了，余额为**80元**的短信肯定要在余额为**30元**的短信之前，否则的话余额先是30元然后又变成80元，这就很奇怪了~

生产环境中，我们并不要求支付成功后，实时的发送余额短信，这时候我们是完全可以利用MQ进行异步解耦的，但是必须要使用**顺序消息**，否则的话就可能出现我上面所诉的情况，出现**线上事故！**

![image-20230629220806996][]

![get_code_YjY2NGM1ZmY2NjQ4ZGNiNGZkMzVmMTBkYTcyOWJiZTAsMTY5Mjg1NDE5MDMzOA][]

--------------------

# RocketMQ顺序消息类型 #

# 全局顺序消息 #

对于指定的一个Topic，所有消息按照严格的先入先出（FIFO）的顺序来发布和消费（单生产者单线程，单消费者单线程）

*  适用场景
 *  适用于性能要求不高，所有的消息严格按照FIFO原则来发布和消费的场景。

# 分区顺序消息 #

对于指定的一个Topic，所有消息根据Sharding Key进行划分到不同队列中，同一个队列内的消息按照严格的先进先出（FIFO）原则进行发布和消费。同一队列内同一Sharding Key的消息保证顺序，不同队列之间的消息顺序不做要求。

*  适用场景
 *  适用于性能要求高，以Sharding Key作为划分字段，在同一个区块中严格地按照先进先出（FIFO）原则进行消息发布和消费的场景。

以RocketMQ中提供的顺序消息案例来看

Producer发送消息的时候自定义了MessageQueueSelector，同时传入的orderId，在选择队列的时候，根据orderId % mqs.size()来选择，这样一来就保证了同一个orderId的消息肯定被发送到同一个msgQueue中，从而保证**分区顺序**

而且Consumer方，需要指定MessageListenerOrderly来消费顺序消息

java复制代码package org.apache.rocketmq.example.ordermessage;
    
    // Producer
    public class Producer {
      public static void main(String[] args) throws UnsupportedEncodingException {
        try {
          DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
          producer.start();
    
          String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
          for (int i = 0; i < 100; i++) {
            int orderId = i % 10;
            Message msg =
              new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,
                          ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
    
            // 自定义MessageQueueSelector来根据arg来选择消息队列
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
              @Override
              public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Integer id = (Integer) arg;
                // 订单号与mqSize取模
                int index = id % mqs.size();
                return mqs.get(index);
              }
            }, orderId);
    
            System.out.printf("%s%n", sendResult);
          }
    
          producer.shutdown();
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
    
    // Consumer
    public class Consumer {
    
      public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
    
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
        consumer.subscribe("TopicTest", "TagA || TagC || TagD");
    
        // 消费者方要使用MessageListenerOrderly来消费顺序消息
        consumer.registerMessageListener(new MessageListenerOrderly() {
          AtomicLong consumeTimes = new AtomicLong(0);
    
          @Override
          public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
            context.setAutoCommit(true);
            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
            this.consumeTimes.incrementAndGet();
            if ((this.consumeTimes.get() % 2) == 0) {
              return ConsumeOrderlyStatus.SUCCESS;
            } else if ((this.consumeTimes.get() % 5) == 0) {
              context.setSuspendCurrentQueueTimeMillis(3000);
              return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
    
            return ConsumeOrderlyStatus.SUCCESS;
          }
        });
    
        consumer.start();
        System.out.printf("Consumer Started.%n");
      }
    
    }

--------------------

# RocketMQ顺序消息原理 #

在生产环境中，我们不建议去实现全局顺序，所以我们下面还是来探究分区顺序消息~

--------------------

# Producer保证顺序 #

在上面的官方案例中，我们提到过：Producer发送消息的时候自定义了MessageQueueSelector，同时传入的orderId，在选择队列的时候，根据orderId % mqs.size()来选择，这样一来就保证了同一个orderId的消息肯定被发送到同一个msgQueue中，从而保证**分区顺序**

参考源码，Producer在发送消息时，确实**回调**了自定义的MessageQueueSelector来选择消息队列

java复制代码private SendResult sendSelectImpl(
      Message msg,
      MessageQueueSelector selector,
      Object arg,
      final CommunicationMode communicationMode,
      final SendCallback sendCallback, final long timeout
    ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
       // ......
    
      // 获取topic发布信息
      TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
      if (topicPublishInfo != null && topicPublishInfo.ok()) {
        MessageQueue mq = null;
        try {
          // 解析出topic对应的队列集合
          List<MessageQueue> messageQueueList =
            mQClientFactory.getMQAdminImpl().parsePublishMessageQueues(topicPublishInfo.getMessageQueueList());
          Message userMessage = MessageAccessor.cloneMessage(msg);
          String userTopic = NamespaceUtil.withoutNamespace(userMessage.getTopic(), mQClientFactory.getClientConfig().getNamespace());
          userMessage.setTopic(userTopic);
    
          // 回调select，选择具体的messageQueue
          mq = mQClientFactory.getClientConfig().queueWithNamespace(selector.select(messageQueueList, userMessage, arg));
        } catch (Throwable e) {
          throw new MQClientException("select message queue threw exception.", e);
        }
    
        // ......
    
        if (mq != null) {
          // 向broker消息
          return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, null, timeout - costTime);
        } else {
          throw new MQClientException("select message queue return null.", null);
        }
      }
      // ......
    }

--------------------

# Consumer保证顺序 #

Consumer启动后，会判断消费者Listener类型

*  如果类型是MessageListenerOrderly表示要进行**顺序消费**，此时使用ConsumeMessageOrderlyService对ConsumeMessageService进行实例化
 *  如果类型是MessageListenerConcurrently表示要进行**并发消费**，此时使用ConsumeMessageConcurrentlyService对ConsumeMessageService进行实例化

java复制代码// org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#start
    public synchronized void start() throws MQClientException {
      switch (this.serviceState) {
        case CREATE_JUST:
          // ......
          
          // 判断消费者Listener类型
          if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
            this.consumeOrderly = true;
            // 说明是顺序消费，使用ConsumeMessageOrderlyService
            this.consumeMessageService =
              new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
          } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
            this.consumeOrderly = false;
            // 并发消费，使用ConsumeMessageConcurrentlyService
            this.consumeMessageService =
              new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
          }
    
          // 启动
          this.consumeMessageService.start();
    
          // ......
      }
    
      // ......
    }

--------------------

# 定时任务对消息队列加锁 #

ConsumeMessageOrderlyService\#start中，如果是处于集群模式下，则会开启一个定时任务，**周期性对消息队列加锁**

java复制代码// org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService#start
    public void start() {
      // 如果是集群模式才加锁
      if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
          @Override
          public void run() {
            try {
              ConsumeMessageOrderlyService.this.lockMQPeriodically();
            } catch (Throwable e) {
              log.error("scheduleAtFixedRate lockMQPeriodically exception", e);
            }
          }
        }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
      }
    }

最终走到RebalanceImpl\#lockAll

1.  从processQueueTable中，映射出broker -> Set<MsgQueue>的关系（map）
2.  以broker维度进行遍历，根据brokerName获取broker信息
3.  对该broker下的所有**消息队列**（即上面map对应的消息队列集合），**批量发送加锁请求**
4.  处理加锁成功、未成功的消息队列

java复制代码// org.apache.rocketmq.client.impl.consumer.RebalanceImpl#lockAll
    public void lockAll() {
      // 映射broker -> Set<MsgQueue>的关系
      HashMap<String, Set<MessageQueue>> brokerMqs = this.buildProcessQueueTableByBrokerName();
    
      // 遍历
      Iterator<Entry<String, Set<MessageQueue>>> it = brokerMqs.entrySet().iterator();
      while (it.hasNext()) {
        Entry<String, Set<MessageQueue>> entry = it.next();
        final String brokerName = entry.getKey();
        final Set<MessageQueue> mqs = entry.getValue();
    
        if (mqs.isEmpty())
          continue;
    
        // 根据brokerName获取broker信息
        FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(brokerName, MixAll.MASTER_ID, true);
        if (findBrokerResult != null) {
    
          // 构建加锁请求
          LockBatchRequestBody requestBody = new LockBatchRequestBody();
          requestBody.setConsumerGroup(this.consumerGroup);
          requestBody.setClientId(this.mQClientFactory.getClientId());
          requestBody.setMqSet(mqs);
    
          try {
            // 批量加锁，会返回加锁成功的消息队列
            Set<MessageQueue> lockOKMQSet =
              this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000);
    
            // 遍历加锁成功的队列
            for (MessageQueue mq : lockOKMQSet) {
              ProcessQueue processQueue = this.processQueueTable.get(mq);
              if (processQueue != null) {
                if (!processQueue.isLocked()) {
                  log.info("the message queue locked OK, Group: {} {}", this.consumerGroup, mq);
                }
    
                // 标记为加锁成功
                processQueue.setLocked(true);
                processQueue.setLastLockTimestamp(System.currentTimeMillis());
              }
            }
    
            // 遍历所有队列，如果不存在于lockOKMQSet中，则表明加速失败
            for (MessageQueue mq : mqs) {
              if (!lockOKMQSet.contains(mq)) {
                ProcessQueue processQueue = this.processQueueTable.get(mq);
                if (processQueue != null) {
                  // 标记为加锁失败
                  processQueue.setLocked(false);
                  log.warn("the message queue locked Failed, Group: {} {}", this.consumerGroup, mq);
                }
              }
            }
          } catch (Exception e) {
            log.error("lockBatchMQ exception, " + mqs, e);
          }
        }
      }
    }

--------------------

# 构造拉取消息请求加锁 #

Consumer构造拉取消息请求逻辑处于RebalanceImpl\#updateProcessQueueTableInRebalance

但由于**重平衡**的机制存在，**当前消息者所属的消息队列**可能存在一定的变动，会被**分配到新的消息队列**，但此时定时任务加锁可能会不够及时，

所以消费者在**构建拉取消息请求前**，会对顺序消息队列**再次检查并加锁**

1.  遍历分配的消息队列（重平衡机制存在，mqSet为新分配的Consumer所属的消息队列）,处理新加入的msgQueue
2.  如果是顺序消息，且当前msgQueue加锁失败，则直接跳过不处理
3.  普通消息 或者 加锁成功的顺序消息，如果不存在于processQueueTable，则为该msgQueue构建新的PullRequest
4.  添加消息拉取请求dispatchPullRequest

java复制代码// org.apache.rocketmq.client.impl.consumer.RebalanceImpl#updateProcessQueueTableInRebalance
    private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet,
                                                       final boolean isOrder) {
      boolean changed = false;
    
      // ......
    
      // 遍历分配的消息队列
      List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
      for (MessageQueue mq : mqSet) {
        // processQueueTable不包含当前mqQueue，说明是新分配的mqQueue
        if (!this.processQueueTable.containsKey(mq)) {
    
          // 如果是顺序消息且加锁失败，则直接跳过不处理
          if (isOrder && !this.lock(mq)) {
            log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
            continue;
          }
    
          // 删除当前消息队列的offSet
          this.removeDirtyOffset(mq);
          // 标记ProcessQueue为加锁成功
          ProcessQueue pq = new ProcessQueue();
          pq.setLocked(true);
    
          long nextOffset = -1L;
          try {
            // 计算新的offSet
            nextOffset = this.computePullFromWhereWithException(mq);
          } catch (Exception e) {
            log.info("doRebalance, {}, compute offset failed, {}", consumerGroup, mq);
            continue;
          }
    
          if (nextOffset >= 0) {
            ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
            if (pre != null) {
              log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
            } else {
              // 说明之前该消息队列不属于该Consumer，则需要构建新的PullRequest
              log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
              PullRequest pullRequest = new PullRequest();
              pullRequest.setConsumerGroup(consumerGroup);
              pullRequest.setNextOffset(nextOffset);
              pullRequest.setMessageQueue(mq);
              pullRequest.setProcessQueue(pq);
              pullRequestList.add(pullRequest);
              changed = true;
            }
          } else {
            log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
          }
        }
      }
    
      // 添加消息拉取请求
      this.dispatchPullRequest(pullRequestList);
    
      return changed;
    }

--------------------

# 消费顺序消息 #

上一步构建好拉取消息请求后，会将请求添加到PullMessageService 的 pullRequestQueue中，同时会启动线程，阻塞从pullRequestQueue获取pullRequest再拉取消息

java复制代码// org.apache.rocketmq.client.impl.consumer.PullMessageService#run
    public void run() {
      log.info(this.getServiceName() + " service started");
    
      while (!this.isStopped()) {
        try {
          PullRequest pullRequest = this.pullRequestQueue.take();
          // 拉取消息
          this.pullMessage(pullRequest);
        } catch (InterruptedException ignored) {
        } catch (Exception e) {
          log.error("Pull Message Service Run Method exception", e);
        }
      }
    
      log.info(this.getServiceName() + " service end");
    }

消息拉取成功后会存在PullCallback的回调，在onSuccess中，则代表拉取成功

*  如果未拉取到消息，则 将拉取请求放入队列再重试
 *  如果拉取到消息，则将消息添加到processQueue，并提交消费请求(submitConsumeRequest)

java复制代码// org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage
    public void pullMessage(final PullRequest pullRequest) {
      // ......
    
      // 拉取消息回调
      PullCallback pullCallback = new PullCallback() {
        @Override
        public void onSuccess(PullResult pullResult) {
          if (pullResult != null) {
            // ......
            
            // 判断拉取结果
            switch (pullResult.getPullStatus()) {
              case FOUND:
                // ......
                long firstMsgOffset = Long.MAX_VALUE;
                
                // 未拉取到消息
                if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
                  // 将拉取请求放入队列再重试
                  DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                } else {
                  // ......
    
                  // 向processQueue添加消息，并提交消费请求
                  boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
                  DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
                    pullResult.getMsgFoundList(),
                    processQueue,
                    pullRequest.getMessageQueue(),
                    dispatchToConsume);
    
                  // ......
                }
                // ......
            }
          }
        }
      };
    
      // .....
    }

顺序消息实现类为ConsumeMessageOrderlyService，通过下面源码可见，即使是顺序消息也是利用线程池进行异步消费，既然这样，那顺序消息如何在多线程环境下保证有序被消费呢？下面接着看~

java复制代码// org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService#submitConsumeRequest
    public void submitConsumeRequest(
      final List<MessageExt> msgs,
      final ProcessQueue processQueue,
      final MessageQueue messageQueue,
      final boolean dispathToConsume) {
      if (dispathToConsume) {
        ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue);
        this.consumeExecutor.submit(consumeRequest);
      }
    }

ConsumeRequest实现了Runnable接口，实现了run方法

1.  processQueue被删除，直接return，不处理
2.  **消息消费队列加锁**，调用fetchLockObject获取对象并使用synchronized加对象锁，保证即使**顺序消息即使是线程池多线程消费**，但是对于**同一个消息队列，只会有一个消费者消费**
3.  如果是广播模式，或者当前的消息队列已经加锁成功（lock字段为true）并且加锁时间**未过期**，才开始对拉取的消息进行消费
4.  执行校验逻辑 集群模式下processQueue未加锁 或者 集群模式下processQueue锁过期，则延时重试加锁，并延时重新消费该processQueue 如果**当前时间距离开始处理的时间超过了最大消费时间**，也延时重新消费该processQueue
5.  批量从processQueue取出消息，**加消息消费锁**，回调Consumer自定义的MessageListenerOrderly进行消费（也就是执行消费的业务代码）

java复制代码// org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService.ConsumeRequest#run
    public void run() {
      // processQueue被删除，直接return
      if (this.processQueue.isDropped()) {
        log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
        return;
      }
    
      // todo: 消息消费队列锁
      final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
      synchronized (objLock) {
        
        // 如果是广播模式，或者当前的消息队列已经加锁成功且加锁时间未过期
        if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
            || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
          final long beginTime = System.currentTimeMillis();
          
          // 开始消费消息
          for (boolean continueConsume = true; continueConsume; ) {
            
            // processQueue被删除，直接跳出循环
            if (this.processQueue.isDropped()) {
              log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
              break;
            }
    
            // 校验: 集群模式下processQueue未加锁
            if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                && !this.processQueue.isLocked()) {
              // 延时加锁重试，并延时重试消费
              ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
              break;
            }
    
            // 校验: 集群模式下processQueue锁过期
            if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                && this.processQueue.isLockExpired()) {
              // 延时加锁重试，并延时重试消费
              ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
              break;
            }
    
            // 如果当前时间距离开始处理的时间超过了最大消费时间,则延时重新消费该processQueue
            long interval = System.currentTimeMillis() - beginTime;
            if (interval > MAX_TIME_CONSUME_CONTINUOUSLY) {
              ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10);
              break;
            }
    
            // 批量消费消息个数
            final int consumeBatchSize =
              ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
    
            // 从processQueue获取消息
            List<MessageExt> msgs = this.processQueue.takeMessages(consumeBatchSize);
            defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
            if (!msgs.isEmpty()) {
              // ......
    
              long beginTimestamp = System.currentTimeMillis();
              ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
              boolean hasException = false;
              try {
                // todo: 消息消费锁
                this.processQueue.getConsumeLock().lock();
                if (this.processQueue.isDropped()) {
                  log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}",
                           this.messageQueue);
                  break;
                }
    
                // todo: messageListener回调
                status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
              } catch (Throwable e) {
                log.warn(String.format("consumeMessage exception: %s Group: %s Msgs: %s MQ: %s",
                                       RemotingHelper.exceptionSimpleDesc(e),
                                       ConsumeMessageOrderlyService.this.consumerGroup,
                                       msgs,
                                       messageQueue), e);
                hasException = true;
              } finally {
                // 解锁
                this.processQueue.getConsumeLock().unlock();
              }
    
              // ......
            } else {
              continueConsume = false;
            }
          }
        } else {
          if (this.processQueue.isDropped()) {
            log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
            return;
          }
    
          ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
        }
      }
    }

--------------------

# 消费时 对消息队列加锁 #

在消费消息时，我们可以看到首先会对**消息队列加锁**

java复制代码// todo: 消息消费队列锁
    final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
    synchronized (objLock) {
      // ......
    }

其本质上是是是利用的ConcurrentMap + synchronized来实现的，map中维护每个消息队列对应的Object对象，再使用synchronized对对象加锁，这样一来，在多线程环境下，也能保持**每个消息队列都是单线程消费**，保证了顺序

java复制代码public class MessageQueueLock {
      
      // 维护messageQueue -> Object
      private ConcurrentMap<MessageQueue, Object> mqLockTable =
        new ConcurrentHashMap<MessageQueue, Object>();
    
      public Object fetchLockObject(final MessageQueue mq) {
        // 获取队列对应的Object
        Object objLock = this.mqLockTable.get(mq);
        if (null == objLock) {
          // 没有则初始化
          objLock = new Object();
          Object prevLock = this.mqLockTable.putIfAbsent(mq, objLock);
          if (prevLock != null) {
            objLock = prevLock;
          }
        }
    
        // 返回对应的Object
        return objLock;
      }
    }

--------------------

# 消费消息加锁 #

在真正消费消息前，会对processQueue加**消费锁**

java复制代码try {
      // todo: 消息消费锁
      this.processQueue.getConsumeLock().lock();
      // 消费消息......
    } catch (Throwable e) {
      // ......
      hasException = true;
    } finally {
      // 解锁
      this.processQueue.getConsumeLock().unlock();
    }

既然已经对**消息队列加锁**了，为什么还要再加**消费锁**？

还是因为重平衡机制的存在

**举例**: 重平衡后，A消费者的某个队列被分配给了B消费者，所以需要把该队列从A消费者所属的消息队列集合中**移除**掉，但可能此时该队列正在被消费，所以就不能被移除。

**所以在消息真正被消费的时候还需要加锁**

**反例**: 如果消息在真正被消费的时候没有加锁，那么就可能出现A消费者正在某个队列的消息，**此时还没有更新offSet**，因为重平衡的存在，导致该队列被分配给B消费者，此时B消费者根据offSet去消费消息，就可能出现**重复消费**的情况。

以下是移除队列的源码

java复制代码// org.apache.rocketmq.client.impl.consumer.RebalancePushImpl#removeUnnecessaryMessageQueue
    public boolean removeUnnecessaryMessageQueue(MessageQueue mq, ProcessQueue pq) {
      this.defaultMQPushConsumerImpl.getOffsetStore().persist(mq);
      this.defaultMQPushConsumerImpl.getOffsetStore().removeOffset(mq);
      
      // 顺序消费，且是集群模式
      if (this.defaultMQPushConsumerImpl.isConsumeOrderly()
          && MessageModel.CLUSTERING.equals(this.defaultMQPushConsumerImpl.messageModel())) {
        try {
          // 尝试获取processQueue的消费锁
          if (pq.getConsumeLock().tryLock(1000, TimeUnit.MILLISECONDS)) {
            try {
              // 成功获取，则才会去延时解开消息队列的锁
              return this.unlockDelay(mq, pq);
            } finally {
              pq.getConsumeLock().unlock();
            }
          } else {
            log.warn("[WRONG]mq is consuming, so can not unlock it, {}. maybe hanged for a while, {}",
                     mq,
                     pq.getTryUnlockTimes());
    
            pq.incTryUnlockTimes();
          }
        } catch (Exception e) {
          log.error("removeUnnecessaryMessageQueue Exception", e);
        }
    
        return false;
      }
      return true;
    }

--------------------

# 总结 #

RocketMQ为了保证消息的顺序性，分别从Producer和Consumer都有着相应的设计~

*  Producer方面，为保证顺序消息，可自定义MessageQueueSelector来选择队列。例: orderId % msgQueueSize，从而可保证同一个orderId的相关消息，会被发送到同一个队列里。
 *  Consumer方面，在整体设计上用了三把锁，来保证消息的顺序消费。 定时任务**周期性对消费者所属队列加锁** **消费时对消息队列加锁** **消息真正被消费前，对processQueue加消费锁**

[image-20230629220806996]: https://img-blog.csdnimg.cn/img_convert/d279bf9d85c541726d6248bdff43fde8.webp?x-oss-process=image/format,png
[get_code_YjY2NGM1ZmY2NjQ4ZGNiNGZkMzVmMTBkYTcyOWJiZTAsMTY5Mjg1NDE5MDMzOA]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=YjY2NGM1ZmY2NjQ4ZGNiNGZkMzVmMTBkYTcyOWJiZTAsMTY5Mjg1NDE5MDMzOA==