详解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>