RocketMQ基础-源码分析

今天药忘吃喽~ 2022-10-31 04:20 278阅读 0赞

# 1.源码仓库 #

[https://github.com/apache/rocketmq][https_github.com_apache_rocketmq]

# 2.NameService启动流程 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70][]

启动类：`org.apache.rocketmq.namesrv.NamesrvStartup`

# 3.NameService路由信息（HashMap） #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 1][]

## 3.1路由数据结构 ##

代码：RouteInfoManager

private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 2][]**   **

**    topicQueueTable：**Topic消息队列路由信息，消息发送时根据路由表进行负载均衡

**    brokerAddrTable：**Broker基础信息，包括brokerName、所属集群名称、主备Broker地址

**    clusterAddrTable：**Broker集群信息，存储集群中所有Broker名称

**    brokerLiveTable：**Broker状态信息，NameServer每次收到心跳包是会替换该信息

**    filterServerTable：**Broker上的FilterServer列表，用于类模式消息过滤。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 3][]

## 3.2Broker路由注册 ##

RocketMQ路由注册是通过Broker与NameServer的心跳功能实现的。Broker启动时向集群中所有的NameServer发送心跳信息，**每隔30s**向集群中所有NameServer发送心跳包，NameServer收到心跳包时会更新brokerLiveTable缓存中BrokerLiveInfo的lastUpdataTimeStamp信息，然后NameServer每隔10s扫描brokerLiveTable，如果**连续120S**没有收到心跳包，NameServer将移除Broker的路由信息同时关闭Socket连接。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 4][]

代码：BrokerController\#start

//注册Broker信息
    this.registerBrokerAll(true, false, true);
    //每隔30s上报Broker信息到NameServer
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    
        @Override
        public void run() {
            try {
                BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
            } catch (Throwable e) {
                log.error("registerBrokerAll Exception", e);
            }
        }
    }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), 
                                                      TimeUnit.MILLISECONDS);

代码：BrokerOuterAPI\#registerBrokerAll

//获得nameServer地址信息
    List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
    //遍历所有nameserver列表
    if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
    
        //封装请求头
        final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
        requestHeader.setBrokerAddr(brokerAddr);
        requestHeader.setBrokerId(brokerId);
        requestHeader.setBrokerName(brokerName);
        requestHeader.setClusterName(clusterName);
        requestHeader.setHaServerAddr(haServerAddr);
        requestHeader.setCompressed(compressed);
    	//封装请求体
        RegisterBrokerBody requestBody = new RegisterBrokerBody();
        requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
        requestBody.setFilterServerList(filterServerList);
        final byte[] body = requestBody.encode(compressed);
        final int bodyCrc32 = UtilAll.crc32(body);
        requestHeader.setBodyCrc32(bodyCrc32);
        final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
        for (final String namesrvAddr : nameServerAddressList) {
            brokerOuterExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        //分别向NameServer注册
                        RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
                        if (result != null) {
                            registerBrokerResultList.add(result);
                        }
    
                        log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
                    } catch (Exception e) {
                        log.warn("registerBroker Exception, {}", namesrvAddr, e);
                    } finally {
                        countDownLatch.countDown();
                    }
                }
            });
        }
    
        try {
            countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
        }
    }

代码：BrokerOutAPI\#registerBroker

if (oneway) {
        try {
            this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
        } catch (RemotingTooMuchRequestException e) {
            // Ignore
        }
        return null;
    }
    RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);

## 3.3NameServic路由注册和心跳检测 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 5][]

代码入口：`org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor`网路处理类解析请求类型，如果请求类型是为***REGISTER\_BROKER***，则将请求转发到`RouteInfoManager#regiesterBroker`

## 3.3 NameService路由删除 ##

`Broker`每隔30s向`NameServer`发送一个心跳包，心跳包包含`BrokerId`，`Broker`地址，`Broker`名称，`Broker`所属集群名称、`Broker`关联的`FilterServer`列表。但是如果`Broker`宕机，`NameServer`无法收到心跳包，此时`NameServer`如何来剔除这些失效的`Broker`呢？`NameServer`会每隔10s扫描`brokerLiveTable`状态表，如果`BrokerLive`的**lastUpdateTimestamp**的时间戳距当前时间超过120s，则认为`Broker`失效，移除该`Broker`，关闭与`Broker`连接，同时更新`topicQueueTable`、`brokerAddrTable`、`brokerLiveTable`、`filterServerTable`。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 6][]

代码入口：代码：NamesrvController\#initialize

# 3.4路由发现 #

RocketMQ路由发现是非实时的，当Topic路由出现变化后，NameServer不会主动推送给客户端，而是由客户端定时拉取主题最新的路由。

*** ***   代码：DefaultRequestProcessor\#getRouteInfoByTopic

public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        final GetRouteInfoRequestHeader requestHeader =
            (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
    	//调用RouteInfoManager的方法,从路由表topicQueueTable、brokerAddrTable、filterServerTable中分别填充TopicRouteData的List<QueueData>、List<BrokerData>、filterServer
        TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());
    	//如果找到主题对应你的路由信息并且该主题为顺序消息，则从NameServer KVConfig中获取关于顺序消息相关的配置填充路由信息
        if (topicRouteData != null) {
            if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {
                String orderTopicConf =
                    this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,
                        requestHeader.getTopic());
                topicRouteData.setOrderTopicConf(orderTopicConf);
            }
    
            byte[] content = topicRouteData.encode();
            response.setBody(content);
            response.setCode(ResponseCode.SUCCESS);
            response.setRemark(null);
            return response;
        }
    
        response.setCode(ResponseCode.TOPIC_NOT_EXIST);
        response.setRemark("No topic route info in name server for the topic: " + requestHeader.getTopic()
            + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
        return response;
    }

# 4.Producer 消息生产者 （客户端）分析 #

## 4.1代码集成和 实现关系 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 7][]    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 8][]    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 9][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 10][]

**   DefaultMQProducer属性说明**

producerGroup：生产者所属组  
    createTopicKey：默认Topic  
    defaultTopicQueueNums：默认主题在每一个Broker队列数量  
    sendMsgTimeout：发送消息默认超时时间，默认3s  
    compressMsgBodyOverHowmuch：消息体超过该值则启用压缩，默认4k  
    retryTimesWhenSendFailed：同步方式发送消息重试次数，默认为2，总共执行3次  
    retryTimesWhenSendAsyncFailed：异步方法发送消息重试次数，默认为2  
    retryAnotherBrokerWhenNotStoreOK：消息重试时选择另外一个Broker时，是否不等待存储结果就返回，默认为false  
    maxMessageSize：允许发送的最大消息长度，默认为4M

## 4.2启动流程 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 11][]

代码入口：DefaultMQProducerImpl\#start

//检查生产者组是否满足要求
    this.checkConfig();
    //更改当前instanceName为进程ID
    if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) {
        this.defaultMQProducer.changeInstanceNameToPID();
    }
    //获得MQ客户端实例
    this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQProducer, rpcHook);

## 4.3消息发送 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 12][]

代码入口：DefaultMQProducerImpl\#send(Message msg)

# 5.消息存储 #

## 5.1核心代码 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 13][]

private final MessageStoreConfig messageStoreConfig;	//消息配置属性
    private final CommitLog commitLog;		//CommitLog文件存储的实现类
    private final ConcurrentMap<String/* topic */, ConcurrentMap<Integer/* queueId */, ConsumeQueue>> consumeQueueTable;	//消息队列存储缓存表,按照消息主题分组
    private final FlushConsumeQueueService flushConsumeQueueService;	//消息队列文件刷盘线程
    private final CleanCommitLogService cleanCommitLogService;	//清除CommitLog文件服务
    private final CleanConsumeQueueService cleanConsumeQueueService;	//清除ConsumerQueue队列文件服务
    private final IndexService indexService;	//索引实现类
    private final AllocateMappedFileService allocateMappedFileService;	//MappedFile分配服务
    private final ReputMessageService reputMessageService;//CommitLog消息分发,根据CommitLog文件构建ConsumerQueue、IndexFile文件
    private final HAService haService;	//存储HA机制
    private final ScheduleMessageService scheduleMessageService;	//消息服务调度线程
    private final StoreStatsService storeStatsService;	//消息存储服务
    private final TransientStorePool transientStorePool;	//消息堆外内存缓存
    private final BrokerStatsManager brokerStatsManager;	//Broker状态管理器
    private final MessageArrivingListener messageArrivingListener;	//消息拉取长轮询模式消息达到监听器
    private final BrokerConfig brokerConfig;	//Broker配置类
    private StoreCheckpoint storeCheckpoint;	//文件刷盘监测点
    private final LinkedList<CommitLogDispatcher> dispatcherList;	//CommitLog文件转发请求

## 5.2消息存储流程 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 14][]

消息存储入口：DefaultMessageStore\#putMessage

## 5.3消息刷盘流程 ##

（1）同步刷盘：消息追加到内存后，立即将数据刷写到磁盘文件

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 15][]

代码入口：CommitLog\#handleDiskFlush

（2）异步刷盘：在消息追加到内存后，立即返回给消息发送端。如果开启transientStorePoolEnable，RocketMQ会单独申请一个与目标物理文件（commitLog）同样大小的堆外内存，该堆外内存将使用内存锁定，确保不会被置换到虚拟内存中去，消息首先追加到堆外内存，然后提交到物理文件的内存映射中，然后刷写到磁盘。如果未开启transientStorePoolEnable，消息直接追加到物理文件直接映射文件中，然后刷写到磁盘中。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 16][]

开启transientStorePoolEnable后异步刷盘步骤:

1.  将消息直接追加到ByteBuffer（堆外内存）
2.  CommitRealTimeService线程每隔200ms将ByteBuffer新追加内容提交到MappedByteBuffer中
3.  MappedByteBuffer在内存中追加提交的内容，wrotePosition指针向后移动
4.  commit操作成功返回，将committedPosition位置恢复
5.  FlushRealTimeService线程默认每500ms将MappedByteBuffer中新追加的内存刷写到磁盘

代码入口：CommitLog$CommitRealTimeService\#run

## 5.4过期文件删除机制 ##

由于RocketMQ操作CommitLog、ConsumerQueue文件是基于内存映射机制并在启动的时候回加载CommitLog、ConsumerQueue目录下的所有文件，为了避免内存与磁盘的浪费，不可能将消息永久存储在消息服务器上，所以要引入一种机制来删除已过期的文件。RocketMQ顺序写CommitLog、ConsumerQueue文件，所有写操作全部落在最后一个CommitLog或者ConsumerQueue文件上，之前的文件在下一个文件创建后将不会再被更新。RocketMQ清除过期文件的方法时：如果当前文件在在一定时间间隔内没有再次被消费，则认为是过期文件，可以被删除，RocketMQ不会关注这个文件上的消息是否全部被消费。默认每个文件的过期时间为72小时，通过在Broker配置文件中设置fileReservedTime来改变过期时间，单位为小时。

删除文件操作的条件

1.  指定删除文件的时间点，RocketMQ通过deleteWhen设置一天的固定时间执行一次删除过期文件操作，默认4点
2.  磁盘空间如果不充足，删除过期文件
3.  预留，手工触发。

代码入口：DefaultMessageStore\#addScheduleTask

# 6.Consumer消息消费者分析 #

消息服务器与消费者之间的消息传递也有两种模式：推模式、拉模式。所谓的拉模式，是消费端主动拉起拉消息请求，而推模式是消息达到消息服务器后，推送给消息消费者。RocketMQ消息推模式的实现基于拉模式

6.1关键类

![20210221201835148.png][]   ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 17][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 18][]

**DefaultMQPushConsumer**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 19][]

//消费者组
    private String consumerGroup;	
    //消息消费模式
    private MessageModel messageModel = MessageModel.CLUSTERING;	
    //指定消费开始偏移量（最大偏移量、最小偏移量、启动时间戳）开始消费
    private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
    //集群模式下的消息队列负载策略
    private AllocateMessageQueueStrategy allocateMessageQueueStrategy;
    //订阅信息
    private Map<String /* topic */, String /* sub expression */> subscription = new HashMap<String, String>();
    //消息业务监听器
    private MessageListener messageListener;
    //消息消费进度存储器
    private OffsetStore offsetStore;
    //消费者最小线程数量
    private int consumeThreadMin = 20;
    //消费者最大线程数量
    private int consumeThreadMax = 20;
    //并发消息消费时处理队列最大跨度
    private int consumeConcurrentlyMaxSpan = 2000;
    //每1000次流控后打印流控日志
    private int pullThresholdForQueue = 1000;
    //推模式下任务间隔时间
    private long pullInterval = 0;
    //推模式下任务拉取的条数,默认32条
    private int pullBatchSize = 32;
    //每次传入MessageListener#consumerMessage中消息的数量
    private int consumeMessageBatchMaxSize = 1;
    //是否每次拉取消息都订阅消息
    private boolean postSubscriptionWhenPull = false;
    //消息重试次数,-1代表16次
    private int maxReconsumeTimes = -1;
    //消息消费超时时间
    private long consumeTimeout = 15;

## 6.1消费者启动流程 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 20][]

代码入口：DefaultMQPushConsumerImpl\#start

## 6.2消费者使用mq提供的客户端拉去消息 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 21][]

(1)PullMessageService实现机制

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 22][]

代码入口：PullMessageService\#run

(2)ProcessQueue实现机制

ProcessQueue是MessageQueue在消费端的重现、快照。PullMessageService从消息服务器默认每次拉取32条消息，按照消息的队列偏移量顺序存放在ProcessQueue中，PullMessageService然后将消息提交到消费者消费线程池，消息成功消费后从ProcessQueue中移除。

(3)消息拉取基本流程

a.客户端发起拉取请求

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 23][]

代码入口：DefaultMQPushConsumerImpl\#pullMessage

b.消息服务端Broker组装消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 24][]

代码入口：PullMessageProcessor\#processRequest

c.消息拉取客户端处理消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 25][]

代码入口：MQClientAPIImpl\#processPullResponse

（4）消息拉取长轮询机制分析

RocketMQ未真正实现消息推模式，而是消费者主动向消息服务器拉取消息，RocketMQ推模式是循环向消息服务端发起消息拉取请求，如果消息消费者向RocketMQ拉取消息时，消息未到达消费队列时，如果不启用长轮询机制，则会在服务端等待shortPollingTimeMills时间后（挂起）再去判断消息是否已经到达指定消息队列，如果消息仍未到达则提示拉取消息客户端PULL—NOT—FOUND（消息不存在）；如果开启长轮询模式，RocketMQ一方面会每隔5s轮询检查一次消息是否可达，同时一有消息达到后立马通知挂起线程再次验证消息是否是自己感兴趣的消息，如果是则从CommitLog文件中提取消息返回给消息拉取客户端，否则直到挂起超时，超时时间由消息拉取方在消息拉取是封装在请求参数中，PUSH模式为15s，PULL模式通过DefaultMQPullConsumer\#setBrokerSuspendMaxTimeMillis设置。RocketMQ通过在Broker客户端配置longPollingEnable为true来开启长轮询模式。

代码入口：PullMessageProcessor\#processRequest

## 6.3**消息队列负载与重新分布机制** ##

RocketMQ消息队列重新分配是由RebalanceService线程来实现。一个MQClientInstance持有一个RebalanceService实现，并随着MQClientInstance的启动而启动。

RocketMQ默认提供5中负载均衡分配算法,如:AllocateMessageQueueAveragely/AllocateMessageQueueAveragelyByCircle

注意：消息队列的分配遵循一个消费者可以分配到多个队列，但同一个消息队列只会分配给一个消费者，故如果出现消费者个数大于消息队列数量，则有些消费者无法消费消息。

代码入口：RebalanceService\#run

## 6.4消息消费过程 ##

PullMessageService负责对消息队列进行消息拉取，从远端服务器拉取消息后将消息存储ProcessQueue消息队列处理队列中，然后调用ConsumeMessageService\#submitConsumeRequest方法进行消息消费，使用线程池来消费消息，确保了消息拉取与消息消费的解耦。ConsumeMessageService支持顺序消息和并发消息，核心类图如下：

![20210221214227172.png][]

代码入口：ConsumeMessageConcurrentlyService\#submitConsumeRequest

## 6.5定时消息机制 ##

定时消息是消息发送到Broker后，并不立即被消费者消费而是要等到特定的时间后才能被消费，RocketMQ并不支持任意的时间精度，如果要支持任意时间精度定时调度，不可避免地需要在Broker层做消息排序，再加上持久化方面的考量，将不可避免的带来巨大的性能消耗，所以RocketMQ只支持特定级别的延迟消息。消息延迟级别在Broker端通过messageDelayLevel配置，默认为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”，delayLevel=1表示延迟消息1s,delayLevel=2表示延迟5s,依次类推。

RocketMQ定时消息实现类为ScheduleMessageService，该类在DefaultMessageStore中创建。通过在DefaultMessageStore中调用load方法加载该类并调用start方法启动。

代码入口：ScheduleMessageService\#load

6.6顺序消息

顺序消息实现类是org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService

代码入口：ConsumeMessageOrderlyService\#start

[https_github.com_apache_rocketmq]: https://github.com/apache/rocketmq
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70]: /images/20221024/d25b99fb56734aea92cdc1a60e07668d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 1]: /images/20221024/83d9782beb4d47a79eea45a111a88e25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 2]: /images/20221024/7d8b5a21f1454054b728bd3915d84e7f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 3]: /images/20221024/b513325f73364915bddee586189d113b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 4]: /images/20221024/bd288a6e128e4f3a8cba42a231131bfe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 5]: /images/20221024/57dbaff4cc9446f0830bcd4cb3edc7e5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 6]: /images/20221024/e38ad492b49f4323adeccc5f36cc12f0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20210221160837461.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20210221160849687.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 9]: /images/20221024/9ee6aa9d1de54c5591f31f917bbcf60e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 10]: /images/20221024/e128a2f35e7d4c2b94fd654b2c327dee.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 11]: /images/20221024/3ed277c56d3349b1aea02fe161336964.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 12]: /images/20221024/7af16e183d5e41f8b6e7458344d76a66.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 13]: /images/20221024/79797c8898bb4ab5ae83ac72000fcb70.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 14]: /images/20221024/4b658adee5bd4388ba6d37a8b99356a3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 15]: /images/20221024/6e029ff15fe1480b8b58ab566a03a35b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 16]: /images/20221024/973f894d3f0a446d879b1cea3bf39d3b.png
[20210221201835148.png]: https://img-blog.csdnimg.cn/20210221201835148.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 17]: https://img-blog.csdnimg.cn/20210221201853984.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 18]: /images/20221024/5cd81060128f4feb8bc52a88fc81094f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 19]: /images/20221024/ee9f3d49a96f45659c12cf60218041b6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 20]: /images/20221024/57d1ccc6326346a09a4f5ec609e7a90c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 21]: /images/20221024/affcbd02faee445cb911b7ebf1814a65.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 22]: /images/20221024/c11ef3f8c81a425d95b38f3afea0fe2f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 23]: /images/20221024/36fe91b0993b4900b4340fddb3795ac0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 24]: /images/20221024/1daec8a46fe44a38af1d33bbbcd610a6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 25]: /images/20221024/6623efdcd62f48f6aeb9e74b1bff6eed.png
[20210221214227172.png]: /images/20221024/41350599101e4103a033b157e48dab36.png