RocketMQ——PullConsumer拉取消息

迈不过友情╰ 2021-09-27 00:46 344阅读 0赞

在RocketMQ中一般有两种获取消息的方式，一个是拉(pull，消费者主动去broker拉取)，一个是推(push，主动推送给消费者)，如下图

![70][]

push-优点：及时性、服务端统一处理实现方便

push-缺点：容易造成堆积、负载性能不可控

pull-优点：获得消息状态方便、负载均衡性能可控

pull-缺点：及时性差

前几篇博文主要以push为例的，本博文以pull为例.

1.Producter

package com.gwd.rocketmq;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
    import com.alibaba.rocketmq.client.producer.SendResult;
    import com.alibaba.rocketmq.common.message.Message;
    
    /** 
    * @FileName Producer.java
    * @Description:
    * @author gu.weidong
    * @version V1.0
    * @createtime 2018年6月25日 上午9:48:37 
    * 修改历史：
    * 时间           作者          版本        描述
    *====================================================  
    *
    */
    public class Producer {
    	public static void main(String[] args) throws MQClientException, InterruptedException {  
            //声明并初始化一个producer  
            //需要一个producer group名字作为构造方法的参数，这里为producer1  
            DefaultMQProducer producer = new DefaultMQProducer("producer1");  
              
            //设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔  
            //NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里  
            producer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");  
       //     producer.setVipChannelEnabled(false);//3.2.6的版本没有该设置，在更新或者最新的版本中务必将其设置为false，否则会有问题  
            producer.setRetryTimesWhenSendFailed(3);
          
            //调用start()方法启动一个producer实例  
            producer.start();  
      
            //发送10条消息到Topic为TopicTest，tag为TagA，消息内容为“Hello RocketMQ”拼接上i的值  
            for (int i = 0; i < 10; i++) {  
                try {  
                    Message msg = new Message("TopicTest",// topic  
                            "TagA",// tag  
                            "i"+i,("Hello RocketMQ " + i).getBytes("utf-8")// body  
                    );  
                      
                    //调用producer的send()方法发送消息  
                    //这里调用的是同步的方式，所以会有返回结果  
                    SendResult sendResult = producer.send(msg);  
                  System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(
                   		 new Date())+","+i);
          //          System.out.println(sendResult.getSendStatus()); //发送结果状态  
                    //打印返回结果，可以看到消息发送的状态以及一些相关信息  
         //           System.out.println(sendResult);  
                } catch (Exception e) {  
                    e.printStackTrace();  
                    Thread.sleep(1000);  
                }  
            }  
      
            //发送完消息之后，调用shutdown()方法关闭producer  
            producer.shutdown();  
        }  
    }

PullConsumer

package com.gwd.rocketmq;
    
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.Set;
    
    import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.client.impl.consumer.PullResultExt;
    import com.alibaba.rocketmq.common.message.MessageExt;
    import com.alibaba.rocketmq.common.message.MessageQueue;
    
    /** 
    * @FileName PullConsumer.java
    * @Description:
    * @author gu.weidong
    * @version V1.0
    * @createtime 2018年6月28日 上午9:44:51 
    * 修改历史：
    * 时间           作者          版本        描述
    *====================================================  
    *
    */
    public class PullConsumer {
    	private static final Map<MessageQueue, Long> offsetTable = new HashMap<MessageQueue, Long>();
    	public static void main(String[] args) throws Exception {
    		DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("pullConsumer");
    		consumer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");
    		consumer.start();
    		try {
    			Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest");
    			for(MessageQueue mq:mqs) {
    				System.out.println("Consume from the queue: " + mq);
    			//	long offset = consumer.fetchConsumeOffset(mq, true);
    			//	PullResultExt pullResult =(PullResultExt)consumer.pull(mq, null, getMessageQueueOffset(mq), 32);
    				//消息未到达默认是阻塞10秒，private long consumerPullTimeoutMillis = 1000 * 10;
    				PullResultExt pullResult =(PullResultExt)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;
                    default:
                        break;
    				}
    			}
    		} catch (MQClientException e) {
    			e.printStackTrace();
    		}
    		
    		
    	}
    	
    	private static void putMessageQueueOffset(MessageQueue mq, long offset) {
            offsetTable.put(mq, offset);
        }
    
        private static long getMessageQueueOffset(MessageQueue mq) {
            Long offset = offsetTable.get(mq);
            if (offset != null)
                return offset;
            return 0;
        }
    }

pullBlockIfNotFound源码：

public PullResult pullBlockIfNotFound(MessageQueue mq, String subExpression, long offset, int maxNums)
                throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
            return this.pullSyncImpl(mq, subExpression, offset, maxNums, true, this.getDefaultMQPullConsumer()
                    .getConsumerPullTimeoutMillis());//不设置超时时间时，默认的是10s
        }

我们选的是同步的方式，需要传的参数：

1、消息队列(通过调用消费者的fetchSubscibeMessageQueue(topic)可以得到相应topic的所需要消息队列) ；

2、需要过滤用的表达式 ；

3、偏移量即消费队列的进度 ；

4、一次取消息的最大值 ；

结果：

![70 1][]

[70]: /images/20210923/6cfa187848314c5ea2d8ff1198bec6a5.png
[70 1]: /images/20210923/8a3c691fdb474ca383e5058707a54186.png