深度挖掘 RocketMQ底层源码「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-上)

Posted 洛神灬殇

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了深度挖掘 RocketMQ底层源码「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-上)相关的知识,希望对你有一定的参考价值。

消息消费者(Consumer)

负责消费消息,一般是后台系统负责异步消费。一个消息消费者会从Broker服务器拉取消息、并将其提供给应用程序。从用户应用的角度而言提供了两种消费形式:拉取式消费、推动式消费。我们接下来主要介绍Pull模式

Pull模式的处理机制

Consumer消费的一种类型,应用通常主动调用Consumer的拉消息方法从Broker服务器拉消息、主动权由应用控制。一旦获取了批量消息,应用就会启动消费过程

在RocketMQ中有两种Pull方式,一种是比较原始Pull Consumer,它不提供相关的订阅方法,需要调用pull方法时指定队列进行拉取,并需要自己更新位点。另一种是Lite Pull Consumer,它提供了Subscribe和Assign两种方式,使用起来更加方便。

Pull模式的使用特点

  • 开发者自己维护OffsetStore。
  • 自己保存消费组的offset,比如存入Redis,或调用MQ接口将其保存在Broker端。自主选择Message Queue和offset进行消息拉取。
  • 用户拉去消息时,需要用户自己来决定拉去哪个队列从哪个offset开始,拉去多少消息。

相比Push的运行特点

与PUSH模式相比,PULL模式需要应用层不间断地进行拉取消息然后再执行消费处理,提高了应用层的编码复杂度,为了Pull方式的编程复杂度,RocketMQ提供了调度消费服务(MQPullConsumerScheduleService),在topic的订阅发送变化(初次订阅或距上次拉取消息超时)就触发PULL方式拉取消息。

DefaultMQPullConsumer

针对于DefaultMQPullConsumer源码流程进行相关的分析,对于Push模式而言,Pull
模式比较适应于客户端拉去的速度由自己进行控制处理。而且实现的原理和复杂程度也简单了很多,我们从实现出发,进行分析对应的实现流程。

DefaultMQPullConsumer的Pull拉取模式的开发案例

指定队列模式消费对应队列的消息

    DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
    consumer.setNamesrvAddr("127.0.0.1:9876");
    consumer.start();
    try 
      MessageQueue mq = new MessageQueue();
      mq.setQueueId(0);
      mq.setTopic("lob");
      mq.setBrokerName("brokerName");
      long offset = 26;
      PullResult pullResult = consumer.pull(mq, "*", offset, 32);
      if (pullResult.getPullStatus().equals(PullStatus.FOUND)) 
        System.out.printf("%s%n", pullResult.getMsgFoundList());
        consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
      
     catch (Exception e) 
      e.printStackTrace();
    
    consumer.shutdown();
  

消费所有队列数据

从所有队列进行选择队列模式,并且存储offset在被本地。

DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("groupName");
consumer.setNamesrvAddr("name-serverl-ip:9876;name-server2-ip:9876");
consumer.start();
Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("order-topic");
for(MessageQueue mq:mqs)
    try 
   // 获取消息的offset,指定从store中获取
   long offset = consumer.fetchConsumeOffset(mq,true);
    while(true)
       PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
       putMessageQueueOffset(mq,pullResult.getNextBeginOffset());
      switch(pullResult.getPullStatus())
         case FOUND:
                  List<MessageExt> messageExtList = pullResult.getMsgFoundList();
                  for (MessageExt m : messageExtList) 
                      System.out.println(new String(m.getBody()));
                   
                break;
        case NO_MATCHED_MSG:
                break;
        case NO_NEW_MSG:
                break;
        case OFFSET_ILLEGAL:
                break;
      
  
 catch (Exception e) 
     e.printStackTrace();
 

consumer.shutdown();

// 保存上次消费的消息下标
private static void putMessageQueueOffset(MessageQueue mq,
      long nextBeginOffset) 
      OFFSE_TABLE.put(mq, nextBeginOffset);


// 获取上次消费的消息的下标
private static Long getMessageQueueOffset(MessageQueue mq) 
      Long offset = OFFSE_TABLE.get(mq);
      if(offset != null)
         return offset;
       
   return 0l;

fetchSubscribeMessageQueues(从指定topic中拉取所有消息队列)

根据Topic获取该Topic的所有消息队列,用于遍历消息队列,从每个消息队列中获取消息,

调用DefaultMQPullConsumer.fetchSubscribeMessageQueues(String topic)方法,根据topic获取对应的MessageQueue(即可被订阅的队列),在该方法中最终通过调用MQAdminImpl.fetchSubscribeMessageQueues(String topic)方法从NameServer获取该topic的MessageQueue。

/**
 * @param topic Topic名称
 * @return 该Topic所有的消息队列
 */
@Override
public Set<MessageQueue> fetchSubscribeMessageQueues(String topic) throws MQClientException 
    return this.defaultMQPullConsumerImpl.fetchSubscribeMessageQueues(withNamespace(topic));

fetchSubscribeMessageQueues底层调用

调用MQClientAPIImpl.getTopicRouteInfoFromNameServer(String topic, long timeoutMillis)方法,其中timeoutMillis=3000,该方法向NameServer发送GET_ROUTEINTO_BY_TOPIC请求码获取topic参数对应的Broker信息和topic配置信息,即TopicRouteData对象;.

 public Set<MessageQueue> fetchSubscribeMessageQueues(String topic) throws MQClientException 
        try 
            TopicRouteData topicRouteData = this.mQClientFactory.getMQClientAPIImpl().getTopicRouteInfoFromNameServer(topic, timeoutMillis);
            if (topicRouteData != null) 
                // 2、遍历topicRouteData
                Set<MessageQueue> mqList = MQClientInstance.topicRouteData2TopicSubscribeInfo(topic, topicRouteData);
                if (!mqList.isEmpty()) 
                    return mqList;
                 else 
                    throw new MQClientException("Can not find Message Queue for this topic, " + topic + " Namesrv return empty", null);
                
            
         catch (Exception e) 
            throw new MQClientException(
                "Can not find Message Queue for this topic, " + topic + FAQUrl.suggestTodo(FAQUrl.MQLIST_NOT_EXIST),
                e);
        
 
        throw new MQClientException("Unknow why, Can not find Message Queue for this topic, " + topic, null);
    
fetchSubscribeMessageQueues底层调用

遍历TopicRouteData对象的QueueData列表中每个QueueData对象,首先判断该QueueData对象是否具有读权限,若有则根据该QueueData对象的readQueueNums值,创建readQueueNums个MessageQueue对象,并构成MessageQueue集合;最后返回给MessageQueue集合

public static Set<MessageQueue> topicRouteData2TopicSubscribeInfo(final String topic, final TopicRouteData route) 
        Set<MessageQueue> mqList = new HashSet<MessageQueue>();
        List<QueueData> qds = route.getQueueDatas();
        for (QueueData qd : qds) 
            if (PermName.isReadable(qd.getPerm())) 
                for (int i = 0; i < qd.getReadQueueNums(); i++) 
                    MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i);
                    mqList.add(mq);
                
            
        
        return mqList;
    
消息的三种拉取模式
同步拉取消息
/**
 * @param mq            消息队列
 * @param subExpression 消息tag过滤表达式
 * @param offset        消费组offset(从哪里开始拉去)
 * @param maxNums       一次最大拉去消息数量
 * @param timeout       超时时间
 * @return 存储了拉取状态以及消息
 */
@Override
public PullResult pull(MessageQueue mq, String subExpression, long offset, int maxNums, long timeout)
        throws MQClientException, RemotingException, MQBrokerException, InterruptedException 
    return this.defaultMQPullConsumerImpl.pull(queueWithNamespace(mq), subExpression, offset, maxNums, timeout);

异步拉取消息
/**
 * @param mq            消息队列
 * @param subExpression 消息tag过滤表达式
 * @param offset        消费组offset(从哪里开始拉去)
 * @param maxNums       一次最大拉去消息数量
 * @param timeout       超时时间
 * @param pullCallback  异步回调函数
 * @param timeout       
 * @throws MQClientException
 * @throws RemotingException
 * @throws InterruptedException
 */
@Override
public void pull(MessageQueue mq, String subExpression, long offset, int maxNums, PullCallback pullCallback,
                 long timeout)
        throws MQClientException, RemotingException, InterruptedException 
    this.defaultMQPullConsumerImpl.pull(queueWithNamespace(mq), subExpression, offset, maxNums, pullCallback, timeout);

同步阻塞拉取消息

拉取消息,若没有找到消息,则阻塞一段时间。通过该方法获取该MessageQueue队列下面从offset位置开始的消息内容。

  • maxNums=32即表示获取的最大消息个数。
  • offset为该MessageQueue对象的开始消费位置,可以调用DefaultMQPullConsumer.fetchConsumeOffset(MessageQueue mq, boolean fromStore)方法获取该MessageQueue队列的消费进度来设定参数offset值该方法最终调用DefaultMQPullConsumerImpl.pullSyncImpl(MessageQueue mq, String subExpression, long offset, int maxNums, boolean block)方法
/**
 * @param mq            消息队列
 * @param subExpression tag过滤
 * @param offset        消费组offset
 * @param maxNums       一次最大拉取数量
 * @return
 */
@Override
public PullResult pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums)
        throws MQClientException, RemotingException, MQBrokerException, InterruptedException 
    return this.defaultMQPullConsumerImpl.pullBlockIfNotFound(queueWithNamespace(mq), subExpression, offset, maxNums);

pullBlockIfNotFound 和 pull区别是: 前者在没有找到消息的时候会阻塞一段时间以便等待后续消息进入,后者则会直接返回 NOT_FOUND 。

维护消息队列的Offset
获取队列的消费Offset
/**
 * @param mq 队列
 * @param fromStore 是否从存储获取,true: 从当前服务器存储中获取,false:从远程broker获取
 * @return 消费offset
 */
@Override
public long fetchConsumeOffset(MessageQueue mq, boolean fromStore) throws MQClientException 
    return this.defaultMQPullConsumerImpl.fetchConsumeOffset(queueWithNamespace(mq), fromStore);

调用DefaultMQPullConsumer.fetchConsumeOffset(MessageQueue mq, boolean fromStore)方法获取MessageQueue队列的消费进度,其中fromStore为false表示从存储端(即Broker端)获取消费进度;若fromStore为true表示从本地内存获取消费进度;

  1. 对于从存储端获取消费进度(即fromStore=true)的情况:
  • 对于LocalFileOffsetStore对象,从本地加载offsets.json文件,然后获取该MessageQueue对象的offset值;

(即fromStore=false)对于RemoteBrokerOffsetStore对象,获取逻辑如下:

  1. 以MessageQueue对象的brokername从MQClientInstance. brokerAddrTable中获取Broker的地址;若没有获取到则立即调用updateTopicRouteInfoFromNameServer方法然后再次获取;

  2. 构造QueryConsumerOffsetRequestHeader对象,其中包括topic、consumerGroup、queueId;然后调用MQClientAPIImpl.queryConsumerOffset (String addr, QueryConsumerOffsetRequestHeader requestHeader, long timeoutMillis)方法向Broker发送QUERY_CONSUMER_OFFSET请求码,获取消费进度Offset;

  3. 用上一步从Broker获取的offset更新本地内存的消费进度列表数据RemoteBrokerOffsetStore.offsetTable:ConcurrentHashMap<MessageQueue, AtomicLong>变量值;

更新消费组Offset

更新消费组的Offset,注意:只会在本地内存中更新,并不会同步到远程Broker.

/**
 * @param mq 消息队列
 * @param offset 消费进度
 */
@Override
public void updateConsumeOffset(MessageQueue mq, long offset) throws MQClientException 
    this.defaultMQPullConsumerImpl.updateConsumeOffset(queueWithNamespace(mq), offset);

以上是关于深度挖掘 RocketMQ底层源码「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-上)的主要内容,如果未能解决你的问题,请参考以下文章

深度挖掘RocketMQ底层源码「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(上篇)

深度挖掘RocketMQ底层源码「底层问题分析系列」深度挖掘RocketMQ底层那些导致消息丢失的汇总盘点透析([REJECTREQUEST]

深度挖掘RocketMQ底层源码「底层问题分析系列」深度挖掘RocketMQ底层那些导致消息丢失的汇总盘点透析([REJECTREQUEST]

深度挖掘RocketMQ底层源码「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行调度的流程(Pull模式)

深度挖掘 RocketMQ底层源码「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-上)

精华推荐 | 深入浅出 RocketMQ原理及实战「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(上篇)