Shulou(Shulou.com)06/01 Report--

This article mainly introduces "how the pushConsumer pull message flow is". In the daily operation, I believe many people have doubts about how the pushConsumer pull message flow is. The editor consulted all kinds of information and sorted out a simple and useful operation method. I hope it will be helpful for you to answer the doubt about "how pushConsumer pull message flow is". Next, please follow the editor to study!

This is a classic RocketMq consumer code to get broker messages asynchronously:

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer ("consumerGroup"); consumer.setNamesrvAddr (Constants.NameServerAddr); consumer.subscribe ("topic01", "*"); consumer.setMessageModel (MessageModel.BROADCASTING); / / broadcast messages. All consumers of the same group, scheduled for topic, can receive messages / / consumer.setMessageModel (MessageModel.CLUSTERING) / / Cluster messages-default (topic in the same group, only one end of the cluster message will receive) consumer.registerMessageListener (new MessageListenerConcurrently () {@ Override public ConsumeConcurrentlyStatus consumeMessage (List list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {for (MessageExt messageExt:list) {System.out.println (new java.lang.String (messageExt.getBody ();} return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}}) Consumer.start ();}

Consumer start () method tracking

1. This.defaultMQPushConsumerImpl.start ()

two。 Just start the switch processing logic in which the serviceState status is CREATE_JUST.

3. First use checkCoing () to check whether each configuration of consumer is configured with ok

4. Then copySubscription () is used to build the conhurrentHashMapInner of the local rebalance based on the subject

5. Then build an Instance of MqClientFactory

6. Build a PullWrapper to register with Broker to filter messages

7. Then obtain the offset according to whether the MessageMode is broadcast mode or cluster mode. (the broadcast mode is obtained from the local store of consumer, while the cluster mode needs to go to broker to request the acquisition)

8. Build a consumeMessageService object based on whether the type of listening message is OrderLy or Concurrently

9. Start the consumerMessageService object you just created and call its start method

10. Register with MqClientFactory Instance instance registerConsumer

11. Change the current serviceState state to Running state

twelve。 Then start to get messages from broker, see the following pushConsumer pull message flow

Introduction to the process of pushConsumer pulling messages

Consumer-DefaultMqPushConsumerImpl uses pullMessage (pullRequest) to pull messages. PullAPIWrapper.pullKernelImpl (passing pullReuest, callback callback and other parameters) according to whether synchronous pullMessageSync or asynchronous pullMessageAsync, the pulled message PullResult is parsed to the TreeMap (offset,messageExt) stored in the ProcessQueue queue.

At this point, the study of "what is the process of pulling messages from pushConsumer" is over. I hope to be able to solve your doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.