pushConsumer拉取消息流程是怎樣的

這篇文章主要介紹“pushConsumer拉取消息流程是怎樣的”，在日常操作中，相信很多人在pushConsumer拉取消息流程是怎樣的問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”pushConsumer拉取消息流程是怎樣的”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

這是一段RocketMq經典的consumer異步獲取broker消息的代碼：

    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");    consumer.setNamesrvAddr(Constants.NameServerAddr);    consumer.subscribe("topic01","*");    consumer.setMessageModel(MessageModel.BROADCASTING);//廣播消息，所有相同組，定于topic的消費端都能收到消息    //consumer.setMessageModel(MessageModel.CLUSTERING);//集群消息--默認（相同組內的topic，集群消息只有一端會接收到）    consumer.registerMessageListener(new MessageListenerConcurrently(){      @Override      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,          ConsumeConcurrentlyContext consumeConcurrentlyContext) {        for (MessageExt messageExt:list){          System.out.println(new java.lang.String(messageExt.getBody()));        }        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;      }    });    consumer.start();  }

consumer start（）方法跟蹤

  1. this.defaultMQPushConsumerImpl.start();

  2. 剛啟動serviceState狀態為 CREATE_JUST，進入這個狀態的switch處理邏輯

3. 先用checkCoing（）檢查consumer的各個配置是否配置ok

4. 然后 copySubscription（）用于根據subject構建本地的rebalance的conhurrentHashMapInner

5. 接著構建MqClientFactory的一個Instance

6. 構建PullWrapper，用于去Broker注冊過濾消息

7. 再根據MessageMode是廣播模式還是集群模式獲取offset。（廣播模式是從consumer本地的store獲取，集群模式則是需要去broker去請求獲取）

8. 根據監聽消息的類型是OrderLy還是Concurrently去構建一個consumeMessageService對象

9.啟動剛才創建的consumerMessageService對象，調用其start方法

10. 使用MqClientFactory Instance實例registerConsumer進行注冊

11. 把當前的serviceState狀態變為Running狀態

12.然后就開始從broker獲取消息，請看下面的pushConsumer拉取消息流程

pushConsumer拉取消息流程介紹 

consumer  --DefaultMqPushConsumerImpl 使用pullMessage（pullRequest）拉取消息，pullAPIWrapper.pullKernelImpl（傳遞pullReuest，回調callback等參數）根據是否同步pullMessageSync還是異步pullMessageAsync， 拉取回來的消息PullResult經過解析處理存放到ProcessQueue 隊列里的TreeMap（offset，messageExt）

到此，關于“pushConsumer拉取消息流程是怎樣的”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！