发布日期: 2023-08-18

文章字数: 6k

阅读时长: 25 分

1 demo

1.1 Producer

根据不同订单id的取模，把不同订单的消息分配到不同的MessageQueue，把相同订单消息分配到相同的MessageQueue。

public class ProducerInOrder {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("OrderProducer");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        String[] tags = new String[]{"TagA", "TagC", "TagD"};
        // 订单列表
        List<Order> orderList = new ProducerInOrder().buildOrders();

        Date date = new Date();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String dateStr = sdf.format(date);
        for (int i = 0; i < orderList.size(); i++) {
            // 加个时间前缀
            String body = dateStr + " Order:" + orderList.get(i);
            Message msg = new Message("PartOrder", tags[i % tags.length], "KEY" + i, body.getBytes());

            SendResult sendResult = producer.send(
                    msg,
                    new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                            //根据订单id选择发送queue
                            Long id = (Long) arg;
                            long index = id % mqs.size();
                            return mqs.get((int) index);
                        }
                    },
                    //订单id
                    orderList.get(i).getOrderId());

            System.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",
                    sendResult.getSendStatus(),
                    sendResult.getMessageQueue().getQueueId(),
                    body));
        }

        producer.shutdown();
    }

    /**
     * 订单
     */
    private static class Order {
        private long orderId;
        private String desc;

        public long getOrderId() {
            return orderId;
        }

        public void setOrderId(long orderId) {
            this.orderId = orderId;
        }

        public String getDesc() {
            return desc;
        }

        public void setDesc(String desc) {
            this.desc = desc;
        }

        @Override
        public String toString() {
            return "Order{" +
                    "orderId=" + orderId +
                    ", desc='" + desc + '\'' +
                    '}';
        }
    }

    /**
     * 生成模拟订单数据  3个订单   每个订单4个状态
     */
    private List<Order> buildOrders() {
        List<Order> orderList = new ArrayList<Order>();

        Order orderDemo = new Order();
        orderDemo.setOrderId(20210406001L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);

        orderDemo = new Order();
        orderDemo.setOrderId(20210406002L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);

        orderDemo = new Order();
        orderDemo.setOrderId(20210406001L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);

        orderDemo = new Order();
        orderDemo.setOrderId(20210406003L);
        orderDemo.setDesc("创建");
        orderList.add(orderDemo);

        orderDemo = new Order();
        orderDemo.setOrderId(20210406002L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);

        orderDemo = new Order();
        orderDemo.setOrderId(20210406003L);
        orderDemo.setDesc("付款");
        orderList.add(orderDemo);

        orderDemo = new Order();
        orderDemo.setOrderId(20210406002L);
        orderDemo.setDesc("推送");
        orderList.add(orderDemo);

        orderDemo = new Order();
        orderDemo.setOrderId(20210406003L);
        orderDemo.setDesc("推送");
        orderList.add(orderDemo);

        orderDemo = new Order();
        orderDemo.setOrderId(20210406002L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);

        orderDemo = new Order();
        orderDemo.setOrderId(20210406001L);
        orderDemo.setDesc("推送");
        orderList.add(orderDemo);

        orderDemo = new Order();
        orderDemo.setOrderId(20210406003L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);

        orderDemo = new Order();
        orderDemo.setOrderId(20210406001L);
        orderDemo.setDesc("完成");
        orderList.add(orderDemo);

        return orderList;
    }
}

1.2 Consumer

public class ConsumerInOrder {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("OrderConsumer");
        consumer.setNamesrvAddr("localhost:9876");
        /*
         * 设置Consumer第一次启动，默认从队列尾部开始消费CONSUME_FROM_LAST_OFFSET<br>
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("PartOrder", "TagA || TagC || TagD");
        consumer.registerMessageListener(new MessageListenerOrderly() {
            Random random = new Random();

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                for (MessageExt msg : msgs) {
                    // 可以看到每个queue有唯一的consume线程来消费, 订单对每个queue(分区)有序
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + "queueId=" + msg.getQueueId() + ", content:" + new String(msg.getBody()));
                }
                try {
                    //模拟业务逻辑处理中...
                    TimeUnit.MILLISECONDS.sleep(random.nextInt(300));
                } catch (Exception e) {
                    e.printStackTrace();
                    //这个点要注意：意思是先等一会，一会儿再处理这批消息，而不是放到重试队列里
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

2 有序

2.1 全局顺序消息

对于指定的同一个topic，为其分配一个消息队列，使用一个生产者，所有消息都按照FIFO的顺序进行发布和消费；
适用于性能要求不高，对数据有严格一致性发布和消费的场景。

全局顺序消息

2.2 分区顺序消息

对于指定的同一个topic，所有消息根据Sharding Key进行区块分区。同一个分区内的消息会严格按照FIFO顺序进行发布和消费。而Sharding Key是顺序消息中用来区分不同区的关键字，和普通消息中key不是同一个概念；
适用于对性能要求高，在同一个区块中要求FIFO进行发布和消费的场景。
例如以订单编号或者订单ID为Sharding Key，在同一个订单内需要严格保证消费顺序。但不同订单中，可能订单A先发布，再发布订单B。在消费是先消费订单B，再消费订单A也是可以的。只要订单内部是顺序执行即可。

部分顺序消息

2.3 如何保证有序

Apache RocketMQ 的消息的顺序性分为两部分，生产顺序性和消费顺序性。

2.3.1 生产顺序性

Apache RocketMQ 通过生产者和服务端的协议保障单个生产者串行地发送消息，并按序存储和持久化。
如需保证消息生产的顺序性，则必须满足以下条件：

单一生产者：消息生产的顺序性仅支持单一生产者，不同生产者分布在不同的系统，即使设置相同的消息组，不同生产者之间产生的消息也无法判定其先后顺序。
串行发送：Apache RocketMQ 生产者客户端支持多线程安全访问，但如果生产者使用多线程并行发送，则不同线程间产生的消息将无法判定其先后顺序。

满足以上条件的生产者，将顺序消息发送至 Apache RocketMQ 后，会保证设置了同一消息组的消息，按照发送顺序存储在同一队列中。服务端顺序存储逻辑如下：

相同消息组的消息按照先后顺序被存储在同一个队列。
不同消息组的消息可以混合在同一个队列中，且不保证连续。
生产顺序性
如上图所示，消息组1和消息组4的消息混合存储在队列1中， Apache RocketMQ保证消息组1中的消息G1-M1、G1-M2、G1-M3是按发送顺序存储，且消息组4的消息G4-M1、G4-M2也是按顺序存储，但消息组1和消息组4中的消息不涉及顺序关系。

2.3.2 消费顺序性

Apache RocketMQ 通过消费者和服务端的协议保障消息消费严格按照存储的先后顺序来处理。

如需保证消息消费的顺序性，则必须满足以下条件：

投递顺序：Apache RocketMQ 通过客户端SDK和服务端通信协议保障消息按照服务端存储顺序投递，但业务方消费消息时需要严格按照接收---处理---应答的语义处理消息，避免因异步处理导致消息乱序。
有限重试：Apache RocketMQ 顺序消息投递仅在重试次数限定范围内，即一条消息如果一直重试失败，超过最大重试次数后将不再重试，跳过这条消息消费，不会一直阻塞后续消息处理。

对于需要严格保证消费顺序的场景，请务必设置合理的重试次数，避免参数不合理导致消息乱序。

3 源码追踪

看消费者，回顾下消费者启动过程，定位到DefaultMQPushConsumerImpl#start()方法，对于并发消费和有序消费，初始化ConsumeMessageService有两种对应的不同类：

有序处理：ConsumeMessageOrderlyService；
并发处理（无序处理）：ConsumeMessageOrderlyService。

怎么决定？由你定义的消费者监听类型决定。一种是MessageListenerOrderly、另一种是MessageListenerConcurrently。

在DefaultMQPushConsumerImpl#start()方法中，对消息监听器类型进行了判断，如果类型是MessageListenerOrderly表示要进行顺序消费，此时使用ConsumeMessageOrderlyService对ConsumeMessageService进行实例化，然后调用它的start方法进行启动：

监听类型决定消费模式

3.1 第一把锁

点进去看看：

public class ConsumeMessageOrderlyService {
    public void start() {
        //只有在集群消费模式下才有效
        if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {
            //周期性（20s）执行加锁
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    ConsumeMessageOrderlyService.this.lockMQPeriodically();
                }
            }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
        }
    }

    public synchronized void lockMQPeriodically() {
        if (!this.stopped) {
            // 进行加锁
            this.defaultMQPushConsumerImpl.getRebalanceImpl().lockAll();
        }
    }
}

为什么集群模式下需要加锁？

因为广播模式下，消息队列会分配给消费组下的每一个消费者，而在集群模式下，一个消息队列同一时刻只能被同一个消费组下的某一个消费者进行，所以在广播模式下不存在竞争关系，也就不需要对消息队列进行加锁，而在集群模式下，有可能因为负载均衡等原因将某一个消息队列分配到了另外一个消费者中，因此在集群模式下就要加锁，当某个消息队列被锁定时，其他的消费者不能进行消费。

这第一把锁，就是向Broker申请消息队列的锁。

public abstract class RebalanceImpl {
    /**
     * 消费者订阅的所有队列与处理队列映射
     */
    protected final ConcurrentMap<MessageQueue, ProcessQueue> processQueueTable = new ConcurrentHashMap<MessageQueue, ProcessQueue>(64);

    /**
     * 第一把锁，MessageQueue锁
     * 在集群模式下，有可能因为负载均衡等原因将某一个消息队列分配到了另外一个消费者中，
     * 因此在集群模式下就要加锁，当某个消息队列被锁定时，其他的消费者不能进行消费。
     */
    public void lockAll() {
        //key为broker名称，value为broker下的消息队列
        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();
            //broker名称
            final String brokerName = entry.getKey();
            //对应的队列集合
            final Set<MessageQueue> mqs = entry.getValue();
            if (mqs.isEmpty()) {
                continue;
            }
            // 根据broker名称获取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());
                        }
                    }
                    // 处理加锁失败的消息队列
                    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);
                }
            }
        }
    }
}

在RebalanceImpl#lockAll方法中，首先从处理队列表中获取当前消费者订阅的所有消息队列MessageQueue信息，返回数据是一个MAP，key为broker名称，value为broker下的消息队列，接着对MAP进行遍历，处理每一个broker下的消息队列：

获取broker名称，根据broker名称查找broker的相关信息；
构建加锁请求，在请求中设置要加锁的消息队列，然后将请求发送给broker，表示要对这些消息队列进行加锁；
加锁请求返回的响应结果中包含了加锁成功的消息队列(可能有失败的)，此时遍历加锁成功的消息队列，将消息队列对应的ProcessQueue中的locked属性置为true，表示该消息队列已加锁成功；
处理加锁失败的消息队列，如果响应中未包含某个消息队列的信息，表示此消息队列加锁失败，需要将其对应的ProcessQueue对象中的locked属性置为false，表示加锁失败。

3.2 第二把锁

有了消息队列，那消费者就要着手处理消息的拉取动作。通过之前的学习，消费者拉取消息，那就要构建拉取请求，然后把获取的消息进行消费处理，反馈结果给broker。

这里有两步动作：

构建拉取请求；
消费消息。

拉取消息请求构建在RebalanceImpl的updateProcessQueueTableInRebalance方法中，拉取消息的响应结果处理在PullCallback的onSuccess方法中，接下来看下顺序消费时在这两个过程中是如何处理的。

3.2.1 构建拉取请求

上面已经知道，在使用顺序消息时，会周期性的对订阅的消息队列进行加锁，不过由于负载均衡等原因，有可能给当前消费者分配新的消息队列，此时可能还未来得及通过定时任务加锁，所以消费者在构建消息拉取请求前会再次进行判断，如果processQueueTable中之前未包含某个消息队列，会先调用lock方法进行加锁，lock方法的实现逻辑与lockAll基本一致，如果加锁成功构建拉取请求进行消息拉取，如果加锁失败，则跳过继续处理下一个消息队列：

public abstract class RebalanceImpl {
    private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet,
                                                       final boolean isOrder) {
        // ...
        List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
        // 遍历队列集合
        for (MessageQueue mq : mqSet) {
            // 如果processQueueTable之前不包含当前的消息队列，尝试加锁
            if (!this.processQueueTable.containsKey(mq)) {
                // 如果是顺序消费，调用lock方法进行加锁，如果加锁失败不往下执行，继续处理下一个消息队列
                if (isOrder && !this.lock(mq)) {
                    log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
                    continue;
                }
                // ... 
                // 如果偏移量大于等于0
                if (nextOffset >= 0) {
                    // 放入处理队列表中
                    ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
                    if (pre != null) {
                        log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
                    } else {
                        // 如果之前不存在，构建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;
    }
    
    public boolean lock(final MessageQueue mq) {
        // 获取broker信息
        FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true);
        if (findBrokerResult != null) {
            // 构建加锁请求
            LockBatchRequestBody requestBody = new LockBatchRequestBody();
            requestBody.setConsumerGroup(this.consumerGroup);
            requestBody.setClientId(this.mQClientFactory.getClientId());
            // 设置要加锁的消息队列
            requestBody.getMqSet().add(mq);

            try {
                // 发送加锁请求
                Set<MessageQueue> lockedMq = this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000);
                for (MessageQueue mmqq : lockedMq) {
                    ProcessQueue processQueue = this.processQueueTable.get(mmqq);
                    // 如果加锁成功设置成功标记
                    if (processQueue != null) {
                        processQueue.setLocked(true);
                        processQueue.setLastLockTimestamp(System.currentTimeMillis());
                    }
                }
                boolean lockOK = lockedMq.contains(mq);
                log.info("the message queue lock {}, {} {}",
                        lockOK ? "OK" : "Failed",
                        this.consumerGroup,
                        mq);
                return lockOK;
            } catch (Exception e) {
                log.error("lockBatchMQ exception, " + mq, e);
            }
        }

        return false;
    }
}

3.2.2 消费消息

在PullCallback的onSuccess方法中可以看到，如果从Broker拉取到消息，会调用ConsumeMessageService#submitConsumeRequest方法将消息提交到ConsumeMessageService中进行消费：

public class DefaultMQPushConsumerImpl implements MQConsumerInner {
    public void pullMessage(final PullRequest pullRequest) {
        // ...
        // 拉取消息回调函数
        PullCallback pullCallback = new PullCallback() {
            @Override
            public void onSuccess(PullResult pullResult) {
                if (pullResult != null) {
                    // 处理拉取结果
                    pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
                            subscriptionData);
                    // 判断拉取结果
                    switch (pullResult.getPullStatus()) {
                        case FOUND:
                            // ...
                            // 如果未拉取到消息
                            if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
                                // 将拉取请求放入到阻塞队列中再进行一次拉取
                                DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                            } else {
                                // ...
                                // 如果拉取到消息，将消息提交到ConsumeMessageService中进行消费
                                DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
                                        pullResult.getMsgFoundList(),
                                        processQueue,
                                        pullRequest.getMessageQueue(),
                                        dispatchToConsume);
                                // ...
                            }
                        // ...
                    }
                }
            }
        };
    }
}

前面知道顺序消费时使用的是ConsumeMessageOrderlyService，首先在ConsumeMessageOrderlyService的构造函数中可以看到初始化了一个消息消费线程池，也就是说顺序消费时也是开启多线程进行消费的：

public class ConsumeMessageOrderlyService implements ConsumeMessageService {
    public ConsumeMessageOrderlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,
        MessageListenerOrderly messageListener) {
        // ...
        // 设置消息消费线程池
        this.consumeExecutor = new ThreadPoolExecutor(
            this.defaultMQPushConsumer.getConsumeThreadMin(),
            this.defaultMQPushConsumer.getConsumeThreadMax(),
            1000 * 60,
            TimeUnit.MILLISECONDS,
            this.consumeRequestQueue,
            new ThreadFactoryImpl(consumeThreadPrefix));
        this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ConsumeMessageScheduledThread_"));
    }
}

接下来看submitConsumeRequest方法，可以看到构建了ConsumeRequest对象，将拉取的消息提交到了消息消费线程池中进行消费：

public class ConsumeMessageOrderlyService implements ConsumeMessageService {
   
    @Override
    public void submitConsumeRequest(
        final List<MessageExt> msgs,
        final ProcessQueue processQueue,
        final MessageQueue messageQueue,
        final boolean dispathToConsume) {
        if (dispathToConsume) {
            // 构建ConsumeRequest
            ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue);
            this.consumeExecutor.submit(consumeRequest);
        }
    }
    
}

ConsumeRequest是ConsumeMessageOrderlyService的内部类，它有两个成员变量，分别为MessageQueue消息队列和它对应的处理队列ProcessQueue对象。
在run方法中，对消息进行消费，处理逻辑如下：

判断ProcessQueue是否被删除，如果被删除终止处理；
调用messageQueueLock#fetchLockObject方法获取消息队列的对象锁（第二把锁），然后使用synchronized进行加锁。
这里加锁的原因是因为：顺序消费使用的是线程池，可以设置多个线程同时进行消费，所以某个线程在进行消息消费的时候要对消息队列加锁，防止其他线程并发消费，破坏消息的顺序性；
如果是广播模式、或者当前的消息队列已经加锁成功（Locked置为true）并且加锁时间未过期，开始对拉取的消息进行遍历：
- 如果是集群模式并且消息队列加锁失败，调用tryLockLaterAndReconsume稍后重新进行加锁；
- 如果是集群模式并且消息队列加锁时间已经过期，调用tryLockLaterAndReconsume稍后重新进行加锁；
- 如果当前时间距离开始处理的时间超过了最大消费时间，调用submitConsumeRequestLater稍后重新进行处理；
- 获取批量消费消息个数，从ProcessQueue获取消息内容，如果消息获取不为空，添加消息消费锁（第三把锁），然后调用messageListener#consumeMessage方法进行消息消费；

public class ConsumeMessageOrderlyService implements ConsumeMessageService {
    class ConsumeRequest implements Runnable {
        // 消息队列对应的处理队列
        private final ProcessQueue processQueue;
        // 消息队列
        private final MessageQueue messageQueue;
        public ConsumeRequest(ProcessQueue processQueue, MessageQueue messageQueue) {
            this.processQueue = processQueue;
            this.messageQueue = messageQueue;
        }

        @Override
        public void run() {
            // 处理队列如果已经被置为删除状态，跳过不进行处理
            if (this.processQueue.isDropped()) {
                log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
                return;
            }
            // 获取消息队列的对象锁
            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;
                        }
                        // 1.如果是集群模式并且processQueue的加锁失败
                        if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                                && !this.processQueue.isLocked()) {
                            log.warn("the message queue not locked, so consume later, {}", this.messageQueue);
                            // 稍后进行加锁
                            ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
                            break;
                        }
                        // 2.如果是集群模式并且消息队列加锁时间已经过期
                        if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                                && this.processQueue.isLockExpired()) {
                            log.warn("the message queue lock expired, so consume later, {}", this.messageQueue);
                            // 稍后进行加锁
                            ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
                            break;
                        }

                        long interval = System.currentTimeMillis() - beginTime;
                        // 3.如果当前时间距离开始处理的时间超过了最大消费时间
                        if (interval > MAX_TIME_CONSUME_CONTINUOUSLY) {
                            // 稍后重新进行处理
                            ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10);
                            break;
                        }
                        // 批量消费消息个数
                        final int consumeBatchSize = ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();
                        // 获取消息内容
                        List<MessageExt> msgs = this.processQueue.takeMessages(consumeBatchSize);
                        defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
                        if (!msgs.isEmpty()) {
                            // ...
                            try {
                                // 加消费锁
                                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;
                                }
                                // 到这才开始消费消息
                                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();
                            }
                            // ...
                            ConsumeMessageOrderlyService.this.getConsumerStatsManager()
                                    .incConsumeRT(ConsumeMessageOrderlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);

                            continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
                        } 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);
                }
            }
        }

    }
}

MessageQueueLock中使用了ConcurrentHashMap存储每个消息队列对应的对象锁，对象锁实际上是一个Object类的对象，从Map中获取消息队列的对象锁时，如果对象锁不存在，则新建一个Object对象，并放入Map集合中：

public class MessageQueueLock {
    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();
            // 加入到Map中
            Object prevLock = this.mqLockTable.putIfAbsent(mq, objLock);
            if (prevLock != null) {
                objLock = prevLock;
            }
        }

        return objLock;
    }
}

3.3 第三把锁

即为消息消费锁。ProcessQueue中持有一个消息消费锁，消费者调用consumeMessage进行消息前，会添加消费锁，上面已经知道在处理拉取到的消息时就已经调用messageQueueLock#fetchLockObject方法获取消息队列的对象锁然后使用synchronized对其加锁，那么为什么在消费之前还要再加一个消费锁呢？

public class ProcessQueue {
    // 消息消费锁
    private final Lock lockConsume = new ReentrantLock();

    public Lock getLockConsume() {
        return lockConsume;
    }
}

先看看哪些地方会调用这个东东：

消费锁调用者

removeUnnecessaryMessageQueue从名字猜测是移除无用的消息队列，其归属于类RebalancePushImpl，而它的这个方法基本就是与负载均衡相关了。

既然这里有移除消息队列的风险，那么消费的时候通过消息队列拉取消息必然要保证该队列可用，即没被删除；而删除的时候，也要保证这个队列没人用了，安全地进行删除动作。所以，上述两个动作都需要对队列先进行加锁，再进行各自的处理。

//RebalancePushImpl
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 {
                //加消费锁
                if (pq.getLockConsume().tryLock(1000, TimeUnit.MILLISECONDS)) {
                    try {
                        return this.unlockDelay(mq, pq);
                    } finally {
                        pq.getLockConsume().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;
    }

不过在消费者在消费消息前已经对队列进行了加锁，负载均衡的时候为什么不使用队列锁而要使用消费锁？

这里应该是为了减小锁的粒度，因为消费者在对消息队列加锁后，还进行了一系列的判断（就是ConsumeRequest中的1-2-3），校验都成功之后从处理队列中获取消息内容，之后才开始消费消息，如果负载均衡使用消息队列锁就要等待整个过程完成才有可能加锁成功，这样显然会降低性能，而如果使用消息消费锁，就可以减少等待的时间，并且消费者在进行消息消费前也会判断ProcessQueue是否被移除，所以只要保证consumeMessage方法在执行的过程中，ProcessQueue不被移除即可。

3.4 总结

目前一共涉及了三把锁，它们分别对应不同的情况：

向Broker申请的消息队列锁(第一把锁)
集群模式下一个消息队列同一时刻只能被同一个消费组下的某一个消费者进行，为了避免负载均衡等原因引起的变动，消费者会向Broker发送请求对消息队列进行加锁，如果加锁成功，记录到消息队列对应的ProcessQueue中的locked变量中，它是boolean类型的：

public class ProcessQueue {
private volatile boolean locked = false;
}

消费者处理拉取消息时的消息队列锁(第二把锁)
消费者在处理拉取到的消息时，由于可以开启多线程进行处理，所以处理消息前通过MessageQueueLock中的mqLockTable获取到了消息队列对应的锁，锁住要处理的消息队列，这里加消息队列锁主要是处理多线程之间的竞争：

public class MessageQueueLock {
  private ConcurrentMap<MessageQueue, Object> mqLockTable = new ConcurrentHashMap<MessageQueue, Object>();
}

消息消费锁(第三把锁)
消费者在调用consumeMessage方法之前会加消费锁，主要是为了避免在消费消息时，由于负载均衡等原因，ProcessQueue被删除：

public class ProcessQueue {
  private final Lock consumeLock = new ReentrantLock();
}

顺序消费“三把锁”

Kezade

https://zyxelva.github.io/posts/3931189064.html

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 Kezade !

RocketMQ源码顺序消息三把锁

【RocketMQ学习】16.SpringBoot整合RocketMQ实战

2023-08-22 RocketMQ学习

RocketMQ实战

【RocketMQ学习】14.源码之延迟消息

2023-08-18 RocketMQ学习

RocketMQ源码延迟消息

【RocketMQ学习】15.源码之顺序消息

1 demo

1.1 Producer

1.2 Consumer

2 有序

2.1 全局顺序消息

2.2 分区顺序消息

2.3 如何保证有序

2.3.1 生产顺序性

2.3.2 消费顺序性

3 源码追踪

3.1 第一把锁

3.2 第二把锁

3.2.1 构建拉取请求

3.2.2 消费消息

3.3 第三把锁

3.4 总结

你的赏识是我前进的动力