//基于訂單號(hào)orderNo進(jìn)行哈希取模發(fā)送訂單消息
Message<Order> message = MessageBuilder.withPayload(order).build();
rocketMQTemplate.syncSendOrderly("ORDER_ADD", message, order.getOrderNo());

private SendResult sendSelectImpl(
        Message msg,
        MessageQueueSelector selector,
        Object arg,
        final CommunicationMode communicationMode,
        final SendCallback sendCallback, final long timeout
    ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
      //......
        if (topicPublishInfo != null && topicPublishInfo.ok()) {
            MessageQueue mq = null;
            try {
             //傳入arg也就是我們的orderNo基于selector算法進(jìn)行哈希取模
                mq = mQClientFactory.getClientConfig().queueWithNamespace(selector.select(messageQueueList, userMessage, arg));
            } catch (Throwable e) {
               //......
            }

          //......
    }

public class SelectMessageQueueByHash implements MessageQueueSelector {

    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
    //數(shù)值哈希運(yùn)算
        int value = arg.hashCode();
      //......
      //結(jié)合隊(duì)列數(shù)取模得到隊(duì)列返回
        value = value % mqs.size();
        return mqs.get(value);
    }
}

@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.groupName}",
        topic = "ORDER_ADD",
        consumeMode = ConsumeMode.ORDERLY//同一個(gè)topic下同一個(gè)隊(duì)列只有一個(gè)消費(fèi)者線程消費(fèi)
        
)
@Slf4j
public class OrderMsgListener implements RocketMQListener<Order> {
    @Override
    public void onMessage(Order order) {
        log.info("收到訂單,訂單信息:[{}]，進(jìn)行積分系統(tǒng)、促銷系統(tǒng)、推送系統(tǒng)業(yè)務(wù)處理.....", JSONUtil.toJsonStr(order));
    }
}

public void run() {
            //......
            //消費(fèi)請求線程消費(fèi)前會(huì)獲取消息隊(duì)列鎖
            final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
            //上鎖
            synchronized (objLock) {
                 //......
                                //將消息發(fā)送給實(shí)現(xiàn)有序監(jiān)聽的監(jiān)聽器線程
                                status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
                            } catch (Throwable e) {
                                 //......
                            } finally {
                                 //......
                            }
  //......
        }

1. 若為同步復(fù)制則master需要復(fù)制到slave節(jié)點(diǎn)后才能返回寫狀態(tài)成功
2. 若配置同步刷盤，還需要基于上述步驟再將數(shù)據(jù)寫入磁盤才能返回寫成功
3. 若是異步刷盤和異步復(fù)制，則消息一到master就直接回復(fù)成功

@Component
@RocketMQMessageListener(consumerGroup = "gourp2", topic = "ORDER_ADD")
public class OrderMqListener2 implements RocketMQListener<Order> {

    private static Logger logger = LoggerFactory.getLogger(OrderMqListener2.class);


    @Override
    public void onMessage(Order order) {

        logger.info("訂閱者2收到消息,訂單信息:[{}]，進(jìn)行新春福利活動(dòng).....", JSON.toJSONString(order));

    }
}

SendResult result = producer.send(message);
                if (!"SEND_OK".equals(result.getSendStatus().name())){
                    logger.warn("消息發(fā)送失敗,執(zhí)行重試的邏輯");
                }

#刷盤方式
#- ASYNC_FLUSH 異步刷盤
#- SYNC_FLUSH 同步刷盤
flushDiskType=ASYNC_FLUSH

consumer.registerMessageListener((List<MessageExt> msgs,
                                          ConsumeConcurrentlyContext context) -> {
            String msg = new String(msgs.stream().findFirst().get().getBody());
            logger.info("消費(fèi)收到消息，消息內(nèi)容={}", msg);
            
            //消費(fèi)完全成功再返回成功狀態(tài)
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.groupName}", topic = "ORDER_ADD")
public class OrderMqListener implements RocketMQListener<Order> {

    private static Logger logger = LoggerFactory.getLogger(OrderMqListener.class);


    @Override
    public void onMessage(Order order) {
        logger.info("消費(fèi)者收到訂單,訂單信息:[{}]，進(jìn)行積分系統(tǒng)、促銷系統(tǒng)、推送系統(tǒng)業(yè)務(wù)處理.....", JSON.toJSONString(order));
        updateOrderFinish(order);
    }

    private void updateOrderFinish(Order order){
        logger.info("執(zhí)行dao層邏輯，將訂單設(shè)置下單完成,無論多少次，執(zhí)行到這個(gè)消費(fèi)邏輯都是將訂單設(shè)置為處理完成");
    }

}

@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.groupName}", topic = "ORDER_ADD")
public class OrderMqListener implements RocketMQListener<Order> {

 //......
    @Override
    public void onMessage(Order order) {
        logger.info("消費(fèi)者收到訂單,訂單信息:[{}]，進(jìn)行積分系統(tǒng)、促銷系統(tǒng)、推送系統(tǒng)業(yè)務(wù)處理.....", JSON.toJSONString(order));
        //消費(fèi)者消費(fèi)時(shí)判斷訂單是否存在，如果存在則直接返回
        if (isExist(order)){
            return;
        }
        updateOrderFinish(order);
    }

   

}

public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
       //......

        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
                || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
           
            //如果大于0則說明是延遲消息
            if (msg.getDelayTimeLevel() > 0) {
                if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                    msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
                }
                //設(shè)置topic設(shè)置為SCHEDULE_TOPIC_XXXX
                topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
                //基于等級(jí)獲取延遲消息隊(duì)列
                queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());

               //......
                //基于上述設(shè)置topic和隊(duì)列信息
                msg.setTopic(topic);
                msg.setQueueId(queueId);
            }
        }

         //......
            //基于零拷貝的方式添加
            result = mappedFile.appendMessage(msg, this.appendMessageCallback);
            //......
    }

public void start() {
        if (started.compareAndSet(false, true)) {
            this.timer = new Timer("ScheduleMessageTimerThread", true);
            //遍歷所有延遲級(jí)別
            for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
                Integer level = entry.getKey();
                Long timeDelay = entry.getValue();
                Long offset = this.offsetTable.get(level);
                if (null == offset) {
                    offset = 0L;
                }
                //設(shè)置timer定時(shí)器
                if (timeDelay != null) {
                    //投遞給定時(shí)器對(duì)應(yīng)等級(jí)的定時(shí)任務(wù)
                    this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
                }
            }
          //......
    }

@Override
        public void run() {
            try {
                if (isStarted()) {
                    this.executeOnTimeup();
                }
            } catch (Exception e) {
              //......
            }
        }


public void executeOnTimeup() {

            //基于topic和隊(duì)列id獲取延遲隊(duì)列
            ConsumeQueue cq =
                ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
                    delayLevel2QueueId(delayLevel));

            long failScheduleOffset = offset;

            if (cq != null) {
                //根據(jù)偏移量獲取有效消息
                SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
                if (bufferCQ != null) {
                    try {
                      //......
                        for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
                            long offsetPy = bufferCQ.getByteBuffer().getLong();
                            int sizePy = bufferCQ.getByteBuffer().getInt();
                            long tagsCode = bufferCQ.getByteBuffer().getLong();

                              //......

                            long now = System.currentTimeMillis();
                            //計(jì)算可消費(fèi)時(shí)間
                            long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);

                            nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);

                            long countdown = deliverTimestamp - now;
                            //如果小于0說明可消費(fèi)
                            if (countdown <= 0) {
                                MessageExt msgExt =
                                    ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(
                                        offsetPy, sizePy);

                                if (msgExt != null) {
                                    try {
                                        //清除延遲級(jí)別恢復(fù)到真正的topic和隊(duì)列id
                                        MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
                                      //......
                                        //放到消息隊(duì)列上
                                        PutMessageResult putMessageResult =
                                            ScheduleMessageService.this.writeMessageStore
                                                .putMessage(msgInner);

                                         //......
                                        } else {
                                           //......
                                    } catch (Exception e) {
                                       //......
                               
                            } else {
                                //......
                            }
                        } // end of for

                         //......
        }

public boolean flush(final int flushLeastPages) {
        boolean result = true;
        //拉取文件處理偏移量信息
        MappedFile mappedFile = this.findMappedFileByOffset(this.flushedWhere, this.flushedWhere == 0);
        if (mappedFile != null) {
            long tmpTimeStamp = mappedFile.getStoreTimestamp();
            //基于mmap零拷貝技術(shù)進(jìn)行刷盤
            int offset = mappedFile.flush(flushLeastPages);
            long where = mappedFile.getFileFromOffset() + offset;
            //如果刷盤后的進(jìn)度和預(yù)期一樣說明刷盤成功
            result = where == this.flushedWhere;
            this.flushedWhere = where;
            if (0 == flushLeastPages) {
                //維護(hù)處理時(shí)間
                this.storeTimestamp = tmpTimeStamp;
            }
        }

        return result;
    }

Message<Order> message = MessageBuilder.withPayload(order).build();
        rocketMQTemplate.syncSendOrderly("ORDER_ADD", message, order.getOrderNo());

@Override
    public void run() {
     //.......

        while (!this.isStopped()) {
           //.......
            //客戶端執(zhí)行負(fù)載均衡
            this.mqClientFactory.doRebalance();
        }

       //.......
    }

public void doRebalance(final boolean isOrder) {
       //......
        if (subTable != null) {
            //遍歷每個(gè)topic
            for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
                final String topic = entry.getKey();
                try {
                    //計(jì)算該topic中當(dāng)前消費(fèi)者要處理的隊(duì)列
                    this.rebalanceByTopic(topic, isOrder);
                } catch (Throwable e) {
                   //......
                }
            }
        }

       //......
    }

private void rebalanceByTopic(final String topic, final boolean isOrder) {
        switch (messageModel) {
            case BROADCASTING: {
               //......
            }
            case CLUSTERING: {
                //根據(jù)主題獲取消息隊(duì)列
                Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
                //根據(jù) topic 與 consumerGroup 獲取所有的 consumerId
                List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
                //......

                if (mqSet != null && cidAll != null) {
                    List<MessageQueue> mqAll = new ArrayList<MessageQueue>();
                    mqAll.addAll(mqSet);
                    //// 排序后才能保證消費(fèi)者負(fù)載策略相對(duì)穩(wěn)定
                    Collections.sort(mqAll);
                    Collections.sort(cidAll);

                    AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;

                    List<MessageQueue> allocateResult = null;
                    try {
                        //按負(fù)載策略進(jìn)行分配，返回當(dāng)前消費(fèi)者實(shí)際訂閱的messageQueue集合
                        allocateResult = strategy.allocate(
                                this.consumerGroup,
                                this.mQClientFactory.getClientId(),
                                mqAll,
                                cidAll);
                    } catch (Throwable e) {
                        log.error("AllocateMessageQueueStrategy.allocate Exception. allocateMessageQueueStrategyName={}", strategy.getName(),
                                e);
                        return;
                    }

                  //......

                    boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);
                    //......
                break;
            }
            default:
                break;
        }
    }

public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
       //......
        //獲取消費(fèi)者id對(duì)應(yīng)的索引
        int index = cidAll.indexOf(currentCID);
        //基于隊(duì)列總數(shù)和客戶端總數(shù)進(jìn)行取模
        int mod = mqAll.size() % cidAll.size();
        /**
         *計(jì)算每個(gè)消費(fèi)者的可消費(fèi)的平均值：
         * 1. 如果消費(fèi)者多于隊(duì)列就取1
         * 2. 如果消費(fèi)者少于隊(duì)列就按照取模結(jié)果來計(jì)算
         */
        
        int averageSize =
            mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()
                + 1 : mqAll.size() / cidAll.size());
        //基于當(dāng)前客戶端的索引定位其處理的隊(duì)列位置
        int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
        //獲取消費(fèi)者的隊(duì)列消費(fèi)范圍
        int range = Math.min(averageSize, mqAll.size() - startIndex);
        //遍歷隊(duì)列存入結(jié)果集
        for (int i = 0; i < range; i++) {
            result.add(mqAll.get((startIndex + i) % mqAll.size()));
        }
        return result;
    }

private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet,
                                                       final boolean isOrder) {
       //......

        List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
        for (MessageQueue mq : mqSet) {
            if (!this.processQueueTable.containsKey(mq)) {
                 //......
                if (nextOffset >= 0) {
                    ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
                    if (pre != null) {
                     //......
                    } 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);
                }
            }
        }
  //存入pullRequestQueue中
        this.dispatchPullRequest(pullRequestList);

        return changed;
    }

@Override
    public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            try {
                //獲取一個(gè)拉消息的請求pullRequest 
                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");
    }

@Override
    public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            try {
                //獲取一個(gè)拉消息的請求
                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");
    }

private RemotingCommand processRequest(final Channel channel, RemotingCommand request, boolean brokerAllowSuspend)
        throws RemotingCommandException {
        RemotingCommand response = RemotingCommand.createResponseCommand(PullMessageResponseHeader.class);
          //.......

            switch (response.getCode()) {
                case ResponseCode.SUCCESS:
     //.......
                    } else {
                        //.......
                    break;
                case ResponseCode.PULL_NOT_FOUND://沒拉取到消息

                    if (brokerAllowSuspend && hasSuspendFlag) {
                        long pollingTimeMills = suspendTimeoutMillisLong;
                        if (!this.brokerController.getBrokerConfig().isLongPollingEnable()) {
                            pollingTimeMills = this.brokerController.getBrokerConfig().getShortPollingTimeMills();
                        }
                        //定位請求的topic以及offset和隊(duì)列id
                        String topic = requestHeader.getTopic();
                        long offset = requestHeader.getQueueOffset();
                        int queueId = requestHeader.getQueueId();
                        //基于上述數(shù)據(jù)生成pullRequest并調(diào)用suspendPullRequest將其hold住
                        PullRequest pullRequest = new PullRequest(request, channel, pollingTimeMills,
                            this.brokerController.getMessageStore().now(), offset, subscriptionData, messageFilter);
                        this.brokerController.getPullRequestHoldService().suspendPullRequest(topic, queueId, pullRequest);
                        response = null;
                        break;
                    }

                case ResponseCode.PULL_RETRY_IMMEDIATELY:
                    break;
                case ResponseCode.PULL_OFFSET_MOVED:
                    //.......
                    break;
                default:
                    assert false;
            }
        } else {
           //.......
        }

         //.......
        return response;
    }

@Override
    public void run() {
        log.info("{} service started", this.getServiceName());
        while (!this.isStopped()) {
            try {
                //等待
                if (this.brokerController.getBrokerConfig().isLongPollingEnable()) {
                    this.waitForRunning(5 * 1000);
                } else {
                    this.waitForRunning(this.brokerController.getBrokerConfig().getShortPollingTimeMills());
                }

                long beginLockTimestamp = this.systemClock.now();
                //基于超時(shí)時(shí)限內(nèi)定時(shí)查看topic中的隊(duì)列是否有新消息，如果有或者超時(shí)則返回
                this.checkHoldRequest();
                long costTime = this.systemClock.now() - beginLockTimestamp;
                if (costTime > 5 * 1000) {
                    log.info("[NOTIFYME] check hold request cost {} ms.", costTime);
                }
            } catch (Throwable e) {
                //......
            }
        }

       //......
    }

public void notifyMessageArriving(final String topic, final int queueId, final long maxOffset, final Long tagsCode,
        //......
        ManyPullRequest mpr = this.pullRequestTable.get(key);
        if (mpr != null) {
            List<PullRequest> requestList = mpr.cloneListAndClear();
            if (requestList != null) {
              //......

                for (PullRequest request : requestList) {
                    long newestOffset = maxOffset;
                    if (newestOffset <= request.getPullFromThisOffset()) {
                        newestOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
                    }
                    //拉取到新消息就返回
                    if (newestOffset > request.getPullFromThisOffset()) {
                        //......

                        if (match) {
                            try {
                                this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(request.getClientChannel(),
                                    request.getRequestCommand());
                            } catch (Throwable e) {
                                log.error("execute request when wakeup failed.", e);
                            }
                            continue;
                        }
                    }
                    //超時(shí)也返回
                    if (System.currentTimeMillis() >= (request.getSuspendTimestamp() + request.getTimeoutMillis())) {
                        try {
                            this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(request.getClientChannel(),
                                request.getRequestCommand());
                        } catch (Throwable e) {
                            log.error("execute request when wakeup failed.", e);
                        }
                        continue;
                    }

                    replayList.add(request);
                }

                //......
            }
        }
    }