消息投递流程

深入理解RocketMQ的消息投递流程，掌握消息发送、消息消费、消息重试等核心机制，理解消息过滤、消息路由等核心功能。

📋 消息投递概览

消息投递是RocketMQ的核心功能，包括消息发送和消息消费两个方向。

🎯 一、消息发送

1.1 发送流程

消息发送流程图：

1.2 DefaultMQProducer

DefaultMQProducer.send：

@Override
public SendResult send(Message msg) throws MQClientException, 
        RemotingException, MQBrokerException, InterruptedException {
    return this.defaultMQProducerImpl.send(msg);
}

🎯 二、消息消费

2.1 消费流程

消息消费流程图：

2.2 DefaultMQPushConsumer

DefaultMQPushConsumer：

public class DefaultMQPushConsumer implements MQConsumer {
    
    @Override
    public void start() throws MQClientException {
        this.defaultMQPushConsumerImpl.start();
    }
}

🎯 三、消息重试

3.1 重试机制

消息重试流程：

• 消费失败：消息消费失败
• 重试队列：消息进入重试队列
• 延迟消费：延迟后重新消费
• 最大重试：达到最大重试次数后进入死信队列

3.2 重试实现

ConsumeMessageConcurrentlyService.processConsumeResult：

public void processConsumeResult(
        final ConsumeConcurrentlyStatus status,
        final ConsumeConcurrentlyContext context,
        final ConsumeRequest consumeRequest) {
    int ackIndex = context.getAckIndex();
    if (consumeRequest.getMsgs().isEmpty()) {
        return;
    }
    
    switch (status) {
        case CONSUME_SUCCESS:
            if (ackIndex >= consumeRequest.getMsgs().size()) {
                ackIndex = consumeRequest.getMsgs().size() - 1;
            }
            int ok = ackIndex + 1;
            int failed = consumeRequest.getMsgs().size() - ok;
            this.getConsumerStatsManager().incConsumeOKTPS(
                    consumerGroup, consumeRequest.getMessageQueue().getTopic(), ok);
            this.getConsumerStatsManager().incConsumeFailedTPS(
                    consumerGroup, consumeRequest.getMessageQueue().getTopic(), failed);
            break;
        case RECONSUME_LATER:
            ackIndex = -1;
            this.getConsumerStatsManager().incConsumeFailedTPS(
                    consumerGroup, consumeRequest.getMessageQueue().getTopic(),
                    consumeRequest.getMsgs().size());
            break;
        default:
            break;
    }
    
    switch (this.defaultMQPushConsumer.getMessageModel()) {
        case BROADCASTING:
            for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
                MessageExt msg = consumeRequest.getMsgs().get(i);
                log.warn("BROADCASTING, the message consume failed, drop it, {}", msg.toString());
            }
            break;
        case CLUSTERING:
            List<MessageExt> msgBackFailed = new ArrayList<MessageExt>(
                    consumeRequest.getMsgs().size());
            for (int i = ackIndex + 1; i < consumeRequest.getMsgs().size(); i++) {
                MessageExt msg = consumeRequest.getMsgs().get(i);
                boolean result = this.sendMessageBack(msg, context);
                if (!result) {
                    msg.setReconsumeTimes(msg.getReconsumeTimes() + 1);
                    msgBackFailed.add(msg);
                }
            }
            
            if (!msgBackFailed.isEmpty()) {
                consumeRequest.getMsgs().removeAll(msgBackFailed);
                this.submitConsumeRequestLater(msgBackFailed, consumeRequest.getProcessQueue(),
                        consumeRequest.getMessageQueue());
            }
            break;
        default:
            break;
    }
    
    long offset = consumeRequest.getProcessQueue().removeMessage(
            consumeRequest.getMsgs());
    if (offset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
        this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(
                consumeRequest.getMessageQueue(), offset, false);
    }
}

📖 学习资源

视频教程

• RocketMQ消息投递源码解析

书籍推荐

• 《RocketMQ技术内幕》 - 丁威
• 《分布式消息中间件实践》 - 倪炜

网站资源

• RocketMQ官方文档

✅ 学习检查清单

• [ ] 已理解消息发送流程
• [ ] 已理解消息消费流程
• [ ] 已理解消息重试机制
• [ ] 已理解消息过滤
• [ ] 已绘制消息投递流程图
• [ ] 已总结核心知识点

---

上一章节：消息存储机制
下一章节：事务消息实现
返回目录：RocketMQ源码深度解析