pushConsumer拉取消息流程是怎樣的

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

這是一段RocketMq經(jīng)典的consumer異步獲取broker消息的代碼：

    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");    consumer.setNamesrvAddr(Constants.NameServerAddr);    consumer.subscribe("topic01","*");    consumer.setMessageModel(MessageModel.BROADCASTING);//廣播消息，所有相同組，定于topic的消費端都能收到消息    //consumer.setMessageModel(MessageModel.CLUSTERING);//集群消息--默認（相同組內(nèi)的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狀態(tài)為 CREATE_JUST，進入這個狀態(tài)的switch處理邏輯

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

4. 然后 copySubscription（）用于根據(jù)subject構(gòu)建本地的rebalance的conhurrentHashMapInner

5. 接著構(gòu)建MqClientFactory的一個Instance

6. 構(gòu)建PullWrapper，用于去Broker注冊過濾消息

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

8. 根據(jù)監(jiān)聽消息的類型是OrderLy還是Concurrently去構(gòu)建一個consumeMessageService對象

9.啟動剛才創(chuàng)建的consumerMessageService對象，調(diào)用其start方法

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

11. 把當前的serviceState狀態(tài)變?yōu)镽unning狀態(tài)

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

pushConsumer拉取消息流程介紹 

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

到此，關(guān)于“pushConsumer拉取消息流程是怎樣的”的學習就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關(guān)知識，請繼續(xù)關(guān)注億速云網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>