| ## DefaultPullConsumer |
| --- |
| ### 类简介 |
| |
| 1. `DefaultMQPullConsumer extends ClientConfig implements MQPullConsumer` |
| |
| 2. `DefaultMQPullConsumer`主动的从Broker拉取消息，主动权由应用控制，可以实现批量的消费消息。Pull方式取消息的过程需要用户自己写，首先通过打算消费的Topic拿到MessageQueue的集合，遍历MessageQueue集合，然后针对每个MessageQueue批量取消息，也可以自定义与控制offset位置。 |
| |
| 3. 优势：consumer可以按需消费，不用担心自己处理能力，而broker堆积消息也会相对简单，无需记录每一个要发送消息的状态，只需要维护所有消息的队列和偏移量就可以了。所以对于慢消费，消息量有限且到来的速度不均匀的情况，pull模式比较合适消息延迟与忙等。 |
| |
| 4. 缺点：由于主动权在消费方，消费方无法及时获取最新的消息。比较适合不及时批处理场景。 |
| |
| ``` java |
| |
| |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| |
| import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer; |
| import org.apache.rocketmq.client.consumer.PullResult; |
| import org.apache.rocketmq.client.exception.MQClientException; |
| import org.apache.rocketmq.common.message.MessageExt; |
| import org.apache.rocketmq.common.message.MessageQueue; |
| |
| public class MQPullConsumer { |
| |
| private static final Map<MessageQueue,Long> OFFSE_TABLE = new HashMap<MessageQueue,Long>(); |
| |
| public static void main(String[] args) throws MQClientException { |
| DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("groupName"); |
| consumer.setNamesrvAddr("127.0.0.1:9876"); |
| consumer.start(); |
| // 从指定topic中拉取所有消息队列 |
| Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("order-topic"); |
| for(MessageQueue mq:mqs){ |
| try { |
| // 获取消息的offset，指定从store中获取 |
| long offset = consumer.fetchConsumeOffset(mq,true); |
| System.out.println("consumer from the queue:"+mq+":"+offset); |
| while(true){ |
| PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, |
| getMessageQueueOffset(mq), 32); |
| putMessageQueueOffset(mq,pullResult.getNextBeginOffset()); |
| switch(pullResult.getPullStatus()){ |
| case FOUND: |
| List<MessageExt> messageExtList = pullResult.getMsgFoundList(); |
| for (MessageExt m : messageExtList) { |
| System.out.println(new String(m.getBody())); |
| } |
| break; |
| case NO_MATCHED_MSG: |
| break; |
| case NO_NEW_MSG: |
| break; |
| case OFFSET_ILLEGAL: |
| break; |
| } |
| } |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| } |
| consumer.shutdown(); |
| } |
| |
| // 保存上次消费的消息下标 |
| private static void putMessageQueueOffset(MessageQueue mq, |
| long nextBeginOffset) { |
| OFFSE_TABLE.put(mq, nextBeginOffset); |
| } |
| |
| // 获取上次消费的消息的下标 |
| private static Long getMessageQueueOffset(MessageQueue mq) { |
| Long offset = OFFSE_TABLE.get(mq); |
| if(offset != null){ |
| return offset; |
| } |
| return 0l; |
| } |
| |
| |
| } |
| ``` |
| |
| |
| |
| ### 字段摘要 |
| |类型|字段名称|描述| |
| |------|-------|-------| |
| |DefaultMQPullConsumerImpl|defaultMQPullConsumerImpl|DefaultMQPullConsumer的内部核心处理默认实现| |
| |String|consumerGroup|消费的唯一分组| |
| |long|brokerSuspendMaxTimeMillis|consumer取连接broker的最大延迟时间，不建议修改| |
| |long|consumerTimeoutMillisWhenSuspend|pull取连接的最大超时时间，必须大于brokerSuspendMaxTimeMillis，不建议修改| |
| |long|consumerPullTimeoutMillis|socket连接的最大超时时间，不建议修改| |
| |String|messageModel|默认cluster模式| |
| |int|messageQueueListener|消息queue监听器，用来获取topic的queue变化| |
| |int|offsetStore|RemoteBrokerOffsetStore 远程与本地offset存储器| |
| |int|registerTopics|注册到该consumer的topic集合| |
| |int|allocateMessageQueueStrategy|consumer的默认获取queue的负载分配策略算法| |
| |
| ### 构造方法摘要 |
| |
| |方法名称|方法描述| |
| |-------|------------| |
| |DefaultMQPullConsumer()|由默认参数值创建一个Pull消费者 | |
| |DefaultMQPullConsumer(final String consumerGroup, RPCHook rpcHook)|使用指定的分组名，hook创建一个消费者| |
| |DefaultMQPullConsumer(final String consumerGroup)|使用指定的分组名消费者| |
| |DefaultMQPullConsumer(RPCHook rpcHook)|使用指定的hook创建一个生产者| |
| |
| |
| ### 使用方法摘要 |
| |
| |返回值|方法名称| 方法描述 | |
| |-------|-------|----------------------------------------------------------------------------------------| |
| |MQAdmin接口method|-------| ------------ | |
| |void|createTopic(String key, String newTopic, int queueNum)| 在broker上创建指定的topic | |
| |void|createTopic(String key, String newTopic, int queueNum, int topicSysFlag)| 在broker上创建指定的topic | |
| |long|earliestMsgStoreTime(MessageQueue mq)| 查询最早的消息存储时间 | |
| |long|maxOffset(MessageQueue mq)| 查询给定消息队列的最大offset | |
| |long|minOffset(MessageQueue mq)| 查询给定消息队列的最小offset | |
| |QueryResult|queryMessage(String topic, String key, int maxNum, long begin, long end)| 按关键字查询消息 | |
| |long|searchOffset(MessageQueue mq, long timestamp)| 查找指定时间的消息队列的物理offset | |
| |MessageExt|viewMessage(String offsetMsgId)| 根据给定的msgId查询消息 | |
| |MessageExt|public MessageExt viewMessage(String topic, String msgId)| 根据给定的msgId查询消息，并指定topic | |
| |MQConsumer接口method|-------| ------------ | |
| |Set<MessageQueue>|fetchSubscribeMessageQueues(String topic)| 根据topic获取订阅的Queue | |
| |void|sendMessageBack(final MessageExt msg, final int delayLevel)| 如果消息出来失败，可以发送回去延迟消费，delayLevel=DelayConf.DELAY_LEVEL | |
| |void|sendMessageBack(final MessageExt msg, final int delayLevel, final String brokerName)| 如果消息出来失败，可以发送回去延迟消费，delayLevel=DelayConf.DELAY_LEVEL | |
| |MQPullConsumer接口method|-------| ------------ | |
| |long|fetchConsumeOffset(MessageQueue mq, boolean fromStore)| 查询给定消息队列的最大offset | |
| |PullResult |pull(final MessageQueue mq, final String subExpression, final long offset,final int maxNums)| 异步拉取制定匹配的消息 | |
| |PullResult| pull(final MessageQueue mq, final String subExpression, final long offset,final int maxNums, final long timeout)| 异步拉取制定匹配的消息 | |
| |PullResult|pull(final MessageQueue mq, final MessageSelector selector, final long offset,final int maxNums)| 异步拉取制定匹配的消息，通过MessageSelector器来过滤消息，参考org.apache.rocketmq.common.filter.ExpressionType | |
| |PullResult|pullBlockIfNotFound(final MessageQueue mq, final String subExpression,final long offset, final int maxNums)| 异步拉取制定匹配的消息，如果没有消息讲block住，并指定超时时间consumerPullTimeoutMillis | |
| |void|pullBlockIfNotFound(final MessageQueue mq, final String subExpression, final long offset,final int maxNums, final PullCallback pullCallback)| 异步拉取制定匹配的消息，如果没有消息讲block住，并指定超时时间consumerPullTimeoutMillis，通过回调pullCallback来消费 | |
| |void|updateConsumeOffset(final MessageQueue mq, final long offset)| 更新指定mq的offset | |
| |long|fetchMessageQueuesInBalance(String topic)| 根据topic获取订阅的Queue(是balance分配后的) | |
| |void|void sendMessageBack(MessageExt msg, int delayLevel, String brokerName, String consumerGroup)| 如果消息出来失败，可以发送回去延迟消费，delayLevel=DelayConf.DELAY_LEVEL，消息可能在同一个consumerGroup消费 | |
| |void|shutdown()| 关闭当前消费者实例并释放相关资源 | |
| |void|start()| 启动消费者 | |
| |
