详解RocketMQ中的consumer

上述就是MQ中有关Consumer的类图，下面来介绍一下每个类

 

 1.MQAdmin：底层类，上篇博客已经提过，就不再此重提

 2.MQConsumer：Consumer公共的接口，常用的方法如下

 

 如果消费失败的话，消息将会返回到broker中，并且延迟一会消费的时间

   void sendMessageBack(final MessageExt msg, final int delayLevel, final String brokerName)  throws RemotingException, MQBrokerException, InterruptedException, MQClientException;

 

 3.MQPushConsumer：Consumer的一种，应用通常向Consumer对象注册一个Listener接口，一旦收到消息，Consumer对象立刻回调Listener接口方法

 

4.MQPullConsumer：Consumer的一种，应用通常主动调用Consumer的拉消息方法从Broker拉消息，主动权由应用控制

 

 在上图中出现了两类的消费者分别是PushConsumer和PullConsumer，下面来看一下

 

 PushConsumer：通过注册监听的方式来消费信息

 

<span style="font-family:Comic Sans MS;font-size:18px;">/**
 * @FileName: Consumer.java
 * @Package:com.test
 * @Description: TODO
 * @author: LUCKY
 * @date:2015年12月28日 下午2:43:23
 * @version V1.0
 */
package com.test;

import java.util.List;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;

/**
 * @ClassName: Consumer
 * @Description: 模拟消费者
 * @author: LUCKY
 * @date:2015年12月28日 下午2:43:23
 */
public class ConsumerTest {

    public static void main(String[] args) {
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("broker-a");
        consumer.setNamesrvAddr("100.66.154.81:9876");
        try {

            // 订阅PushTopic下Tag为push的消息,都订阅消息
            consumer.subscribe("PushTopic", "push");

            // 程序第一次启动从消息队列头获取数据
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //可以修改每次消费消息的数量，默认设置是每次消费一条
            // consumer.setConsumeMessageBatchMaxSize(10);

            //注册消费的监听
            consumer.registerMessageListener(new MessageListenerConcurrently() {
               //在此监听中消费信息，并返回消费的状态信息
                public ConsumeConcurrentlyStatus consumeMessage(
                        List<MessageExt> msgs,
                        ConsumeConcurrentlyContext context) {

                    // msgs中只收集同一个topic，同一个tag，并且key相同的message
                    // 会把不同的消息分别放置到不同的队列中
                    for(Message msg:msgs){

                        System.out.println(new String(msg.getBody()));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            consumer.start();
            Thread.sleep(5000);
            //5秒后挂载消费端消费
            consumer.suspend();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
</span>

 

 

 PullConsumer：通过拉去的方式来消费消息

 

<span style="font-family:Comic Sans MS;font-size:18px;">/**
 * @FileName: Consumer.java
 * @Package:com.test
 * @Description: TODO
 * @author: LUCKY
 * @date:2015年12月28日 下午2:43:23
 * @version V1.0
 */
package com.test;

import java.util.Set;

import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
import com.alibaba.rocketmq.client.consumer.MessageQueueListener;
import com.alibaba.rocketmq.common.message.MessageQueue;

/**
 * @ClassName: Consumer
 * @Description: 模拟消费者
 * @author: LUCKY
 * @date:2015年12月28日 下午2:43:23
 */
public class ConsumerPullTest {

    public static void main(String[] args) {
        DefaultMQPullConsumer consumer=new DefaultMQPullConsumer();
        consumer.setNamesrvAddr("100.66.154.81:9876");
       consumer.setConsumerGroup("broker");
        try {
            consumer.start();
        Set<MessageQueue> messageQueues=  consumer.fetchSubscribeMessageQueues("PushTopic");

        for(MessageQueue messageQueue:messageQueues){

            System.out.println(messageQueue.getTopic());
        }


        //消息队列的监听
        consumer.registerMessageQueueListener("", new MessageQueueListener() {

            @Override
            //消息队列有改变，就会触发
            public void messageQueueChanged(String topic, Set<MessageQueue> mqAll,
                    Set<MessageQueue> mqDivided) {
                // TODO Auto-generated method stub

            }
        });


        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
</span>

一般在应用中都会采用push的方法来自动的消费信息