面试官：RocketMQ 的推模式和拉模式有什么区别？

大家好，我是君哥。,RocketMQ 消息消费有两种模式，PULL 和 PUSH,今天我们来看一下这两种模式有什么区别。,首先看一段 RocketMQ 推模式的一个官方示例：,消费者会定义一个消息监听器，并且把这个监听器注册到 DefaultMQPushConsumer，同时也会注册到 DefaultMQPushConsumerIm-pl，当拉取到消息时，就会使用这个监听器来处理消息。那这个监听器是什么时候调用呢?看下面的 UML 类图：,消费者真正拉取请求的类是 DefaultMQPush-ConsumerImpl，这个类的 pullMessage 方法调用了 PullAPIWrapper 的 pullKernelImpl 方法，这个方法有一个参数是回调函数 Pull-Callback，当 PULL 状态是 PullStatus.FOU-ND 时，代表拉取消息成功，处理逻辑如下：,这个处理逻辑调用了 ConsumeMessage-Service 类的 submitConsumeRequest 方法，我们看一下并发消费消息的处理逻辑，代码如下：,ConsumeRequest 类是一个线程类，run 方法里面调用了消费者定义的消息处理方法，代码如下：,下面以并发消费方式下的同步拉取消息为例总结一下消费者消息处理过程：,下面是来自官方的一段 PULL 模式拉取消息的代码：,这里我们看到，PULL 模式需要在处理逻辑里不停的去拉取消息，比如上面代码中写了一个死循环。那 PULL 模式中 poll 函数是怎么实现的呢?我们看下面的 UML 类图：,跟踪源码可以看到，消息拉取最终是从 DefaultLitePullConsumerImpl 类中的一个 LinkedBlockingQueue 上面拉取。那消息是什么时候 put 到 LinkedBlockingQueue 呢?,官方拉取消息的代码中有一个 subscribe 方法订阅了 Topic，这里相关的 UML 类图如下：,这个 subscribe 方法最终调用了 DefaultLite-PullConsumerImpl 类的 subscribe，代码如下：,这里给 DefaultLitePullConsumer 类的 messageQueueListener 这个监听器进行了赋值。当监听器监听到 MessageQueue 发送变化时，就会启动消息拉取消息的线程 Pull-TaskImpl，代码如下：,拉取消息成功后，调用 submitConsume-Request 方法把拉取到的消息放到 consumeRequestCache，然后启动下一次拉取。,这样就清除了示例代码中 poll 消息的逻辑，那还有一个问题，监听器是什么时候触发监听事件呢?,在消费者启动时，会启动 RebalanceService 这个线程，这个线程的 run 方法如下：,下面的 UML 类图显示了 doRebalance 方法的调用关系：,可以看到最终调用了 最终调用了 Rebalance-LitePullImpl 的 messageQueueChanged 方法，代码如下：,这里最终触发了监听器。,下面以并发消费方式下的同步拉取消息为例总结一下消费者消息处理过程：,通过本文的讲解，可以看到 PUSH 模式和 PULL 模式本质上都是客户端主动拉取，RocketMQ并没有真正实现 Broker 推送消息的 PUSH 模式。RocketMQ 中 PULL 模式和 PUSH 模式的区别如下：,PULL 模式是从 Broker 拉取消息后放入缓存，然后消费端不停地从缓存取出消息来执行客户端定义的处理逻辑，而 PUSH 模式是在死循环中不停的从 Broker 拉取消息，拉取到后调用回调函数进行处理，回调函数中调用客户端定义的处理逻辑。,PUSH 模式拉取消息依赖死循环来不停唤起业务，而 PULL 模式拉取消息是通过 MessageQueue 监听器来触发消息拉取线程，消息拉取线程会在拉取完一次后接着下一次拉取。