RocketMQ 消息消费有两种模式PULL 和 PUSH,今天我们来看一下这两种模式有什么区别。PUSH 模式首先看一段 RocketMQ 推模式的一个官方示例public static void main(String[] args) throws InterruptedException, MQClientException { Tracer tracer initTracer(); DefaultMQPushConsumer consumer new DefaultMQPushConsumer(CID_JODIE_1); consumer.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(new ConsumeMessageOpenTracingHookImpl(tracer)); consumer.subscribe(TopicTest, *); consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); consumer.setConsumeTimestamp(20181109221800); consumer.registerMessageListener(new MessageListenerConcurrently() { Override public ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs, ConsumeConcurrentlyContext context) { System.out.printf(%s Receive New Messages: %s %n, Thread.currentThread().getName(), msgs); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); consumer.start(); System.out.printf(Consumer Started.%n); }消费者会定义一个消息监听器并且把这个监听器注册到 DefaultMQPushConsumer同时也会注册到 DefaultMQPushConsumerIm-pl当拉取到消息时就会使用这个监听器来处理消息。那这个监听器是什么时候调用呢看下面的 UML 类图消费者真正拉取请求的类是 DefaultMQPush-ConsumerImpl这个类的 pullMessage 方法调用了 PullAPIWrapper 的 pullKernelImpl 方法这个方法有一个参数是回调函数 Pull-Callback当 PULL 状态是 PullStatus.FOU-ND 时代表拉取消息成功处理逻辑如下PullCallback pullCallback new PullCallback() { Override public void onSuccess(PullResult pullResult) { if (pullResult ! null) { pullResult DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult, subscriptionData); switch (pullResult.getPullStatus()) { case FOUND: //省略部分逻辑 DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest( pullResult.getMsgFoundList(), processQueue, pullRequest.getMessageQueue(), dispatchToConsume); //省略部分逻辑 break; //省略其他case default: break; } } } Override public void onException(Throwable e) { //省略 } };这个处理逻辑调用了 ConsumeMessage-Service 类的 submitConsumeRequest 方法我们看一下并发消费消息的处理逻辑代码如下public void submitConsumeRequest( final ListMessageExt msgs, final ProcessQueue processQueue, final MessageQueue messageQueue, final boolean dispatchToConsume) { final int consumeBatchSize this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize(); if (msgs.size() consumeBatchSize) { ConsumeRequest consumeRequest new ConsumeRequest(msgs, processQueue, messageQueue); try { this.consumeExecutor.submit(consumeRequest); } catch (RejectedExecutionException e) { this.submitConsumeRequestLater(consumeRequest); } } else { //分批处理跟上面逻辑一致 }ConsumeRequest 类是一个线程类run 方法里面调用了消费者定义的消息处理方法代码如下public void run() { //省略逻辑 MessageListenerConcurrently listener ConsumeMessageConcurrentlyService.this.messageListener; //省略逻辑 try { //调用消费方法 status listener.consumeMessage(Collections.unmodifiableList(msgs), context); } catch (Throwable e) { //省略逻辑 } //省略逻辑 }下面以并发消费方式下的同步拉取消息为例总结一下消费者消息处理过程在 MessageListenerConcurrently 中定义消费者处理逻辑消费者启动时注册到 DefaultMQPushConsumer 和 DefaultMQ-PushConsumerImpl消费者启动时启动消费拉取线程 PullMessageService里面死循环不停地从 Broker 拉取消息。这里调用了 DefaultMQPushConsumerImpl 类的 pullMessage 方法DefaultMQPushConsumerImpl 类的 pullMessage 方法调用 PullAPIWrapper 的 pullKernelImpl 方法真正去发送 PULL 请求并传入 PullCallback 的 回调函数拉取到消息后调用 PullCallback 的 onSuccess 方法处理结果这里调用了 ConsumeMessageConcurrentlyService 的 submitConsumeRequest 方法里面用 ConsumeRequest 线程来处理拉取到的消息ConsumeRequest 处理消息时调用了消费端定义的消费逻辑也就是 Message-ListenerConcurrently 的 consumeMessage 方法。PULL 模式下面是来自官方的一段 PULL 模式拉取消息的代码DefaultLitePullConsumer litePullConsumer new DefaultLitePullConsumer(lite_pull_consumer_test); litePullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); litePullConsumer.subscribe(TopicTest, *); litePullConsumer.start(); try { while (running) { ListMessageExt messageExts litePullConsumer.poll(); System.out.printf(%s%n, messageExts); } } finally { litePullConsumer.shutdown(); }这里我们看到PULL 模式需要在处理逻辑里不停的去拉取消息比如上面代码中写了一个死循环。那 PULL 模式中 poll 函数是怎么实现的呢我们看下面的 UML 类图跟踪源码可以看到消息拉取最终是从 DefaultLitePullConsumerImpl 类中的一个 LinkedBlockingQueue 上面拉取。那消息是什么时候 put 到 LinkedBlockingQueue 呢官方拉取消息的代码中有一个 subscribe 方法订阅了 Topic这里相关的 UML 类图如下这个 subscribe 方法最终调用了 DefaultLite-PullConsumerImpl 类的 subscribe代码如下public synchronized void subscribe(String topic, String subExpression) throws MQClientException { try { //省略逻辑 this.defaultLitePullConsumer.setMessageQueueListener(new MessageQueueListenerImpl()); assignedMessageQueue.setRebalanceImpl(this.rebalanceImpl); //省略逻辑 } catch (Exception e) { throw new MQClientException(subscribe exception, e); } }这里给 DefaultLitePullConsumer 类的 messageQueueListener 这个监听器进行了赋值。当监听器监听到 MessageQueue 发送变化时就会启动消息拉取消息的线程 Pull-TaskImpl代码如下public void run() { //省略部分逻辑 if (!this.isCancelled()) { long pullDelayTimeMills 0; try { PullResult pullResult pull(messageQueue, subscriptionData, offset, defaultLitePullConsumer.getPullBatchSize()); switch (pullResult.getPullStatus()) { case FOUND: final Object objLock messageQueueLock.fetchLockObject(messageQueue); synchronized (objLock) { if (pullResult.getMsgFoundList() ! null !pullResult.getMsgFoundList().isEmpty() assignedMessageQueue.getSeekOffset(messageQueue) -1) { processQueue.putMessage(pullResult.getMsgFoundList()); submitConsumeRequest(new ConsumeRequest(pullResult.getMsgFoundList(), messageQueue, processQueue)); } } break; //省略其他 case } } //省略 catch if (!this.isCancelled()) { //启动下一次拉取 scheduledThreadPoolExecutor.schedule(this, pullDelayTimeMills, TimeUnit.MILLISECONDS); } else { log.warn(The Pull Task is cancelled after doPullTask, {}, messageQueue); } } }拉取消息成功后调用 submitConsume-Request 方法把拉取到的消息放到 consumeRequestCache然后启动下一次拉取。这样就清除了示例代码中 poll 消息的逻辑那还有一个问题监听器是什么时候触发监听事件呢在消费者启动时会启动 RebalanceService 这个线程这个线程的 run 方法如下public void run() { while (!this.isStopped()) { this.waitForRunning(waitInterval); this.mqClientFactory.doRebalance(); } }下面的 UML 类图显示了 doRebalance 方法的调用关系可以看到最终调用了 最终调用了 Rebalance-LitePullImpl 的 messageQueueChanged 方法代码如下public void messageQueueChanged(String topic, SetMessageQueue mqAll, SetMessageQueue mqDivided) { MessageQueueListener messageQueueListener this.litePullConsumerImpl.getDefaultLitePullConsumer().getMessageQueueListener(); if (messageQueueListener ! null) { try { messageQueueListener.messageQueueChanged(topic, mqAll, mqDivided); } catch (Throwable e) { log.error(messageQueueChanged exception, e); } } }这里最终触发了监听器。下面以并发消费方式下的同步拉取消息为例总结一下消费者消息处理过程消费者启动向 DefaultLitePullConsumer 订阅了 Topic这个订阅过程会向 DefaultLitePullConsumer 注册一个监听器消费者启动过程中会启动 Message-Queue 重平衡线程 Rebalance-Service当重平衡过程发现 ProcessQueueTable 发生变化时启动消息拉取线程消息拉取线程拉取到消息后把消息放到 consumeRequestCache然后进行下一次拉取消费者启动后不停地从 consumeReq-uestCache 拉取消息进行处理。总结通过本文的讲解可以看到 PUSH 模式和 PULL 模式本质上都是客户端主动拉取RocketMQ并没有真正实现 Broker 推送消息的 PUSH 模式。RocketMQ 中 PULL 模式和 PUSH 模式的区别如下PULL 模式是从 Broker 拉取消息后放入缓存然后消费端不停地从缓存取出消息来执行客户端定义的处理逻辑而 PUSH 模式是在死循环中不停的从 Broker 拉取消息拉取到后调用回调函数进行处理回调函数中调用客户端定义的处理逻辑PUSH 模式拉取消息依赖死循环来不停唤起业务而 PULL 模式拉取消息是通过 MessageQueue 监听器来触发消息拉取线程消息拉取线程会在拉取完一次后接着下一次拉取。2026年最新版百万八股面试宝典《Java》《大模型》《大并发》《大数据》本题库由资深讲师联合大厂面试官精心打造专为高级IT岗位设计【点击领取-百万八股面试宝典】面试突击体系相关Java企业架构体系相关AI大模型相关HR 面试软技能设计模式相关并发编程相关网络IO与Netty相关互联网三高项目相关亿级流量多级缓存相关数据结构算法相关分布式相关分布式锁相关分布式ID相关核心源码相关大厂线上故障分析相关Docker相关Dubbo相关ElasticSearch相关GoLang相关Java基础核心JVM核心相关JVM调优底层相关Kafka相关Kubernetes相关Linux相关MongoDB相关MQ相关MySQL相关MyBatis相关MyBatisPlus相关Postgresql相关RabbitMQ相关RocketMQ相关Redis相关ShardingSphere相关Shiro相关SkyWalking相关Solr相关Spring 相关SpringBoot相关SpringCloud相关SpringMVC源码相关SpringSecurity相关Spring WebFlux相关领域驱动设计DDD相关任务调度Airflow相关Zookeeper相关HDFS面试题MapReduce面试题Yarn面试题Hive相关SQL面试题相关Scala面试题Spark面试题Flink相关Doris相关Tomcat源码相关网络安全相关运维/云原生相关