PushConsumer与PullConsumer的区别.pdf

PushConsumer 与PullConsumer 的区别 概念上来说PushConsumer 应该是推给消费者，不考虑消费者的消费 能力，PullConsumer 则是消费者主动去拉队列中的消息。 但是RocketMq 与Kafka 的实现方式不太一样，RocketMQ 的 Consumer 都是从 Broker 拉消息来消费，但是为了能做到实时收消 息，RocketMQ 使用长轮询方式，可以保证消息实时性同 Push 方式一 致。返种长轮询方式类似亍 Web QQ 收収消息机制。 PushConsumer 的consumeFromWhere 参数可以指定在消费者第一次启 动时从哪个位置开始消费，默认情况下CONSUME_FROM_LAST_OFFSET 从 队列最后开始消费，也就是不消费之前的数据。后续启动会从上次消 费的位置开始消费，该参数不再起作用，但是需要注意的是重启 Consumer 有可能造成重复消费数据，因为消费的位置信息并不是实时 保存的，这点非常重要，在业务系统上就要防止重复数据提交，持久 化消费的位置信息间隔时间可以通过参数persistConsumerOffsetInterval 配 置，默认是5 秒，但我自己测试发现似乎无效。 使用PullConsumer 则可以指定某个Offset 开始Pull 消息。 测试代码 import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer; import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrently Context ; import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyS tatus ; import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcu rrently ; import com.alibaba.rocketmq.client.exception.MQClientException ; import mon.consumer.ConsumeFromWhere ; import mon.message.MessageExt ; import tocol.heartbeat.MessageModel ; import java.util.List ; import java.util.concurrent.TimeUnit ; /** * Created by Administrator on 2016/1/6 0006. */ public class PushConsumerTest { static int i = 0 ; /** * 当前例子是PushConsumer 用法，使用方式给用户感觉是消息从 RocketMQ 服务器推到了应用客户端。br * 但是实际PushConsumer 内部是使用长轮询Pull 方式从Broker 拉消息， 然后再回调用户Listener 方法br */ public static void main(String[] args) throws InterruptedException, MQClientException { /** * 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局 对象或者单例br * 注意：ConsumerGroupName 需要由应用来保证唯一 */ final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer( TestGroup) ; // consumer.setInstanceName(TestConsumer); consumer.setNamesrvAddr( ipAddress:9876); consumer.setConsumeThreadMin( 1000); consumer.setConsumeThreadMax( 1000); /** * 订阅指定topic