詳解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的方法來自動的消費信息