RoceketMQ Push模式简单样例
public static void main(String[] args) throws InterruptedException, MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1"); //指定Consumer Group
consumer.subscribe("Jodie_topic_1023", "*"); //指定要消费的topic
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.registerMessageListener(new MessageListenerConcurrently() { //该topic的任何一个queue有新消息，该回调回被调用
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.printf("Consumer Started.%n");
}
RoceketMQ Pull模式简单样例
public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5"); //指定Consumer Group
consumer.start();
Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest1"); //获取一个topic的所有MessageQueue
for (MessageQueue mq : mqs) {
System.out.printf("Consume from the queue: " + mq + "%n");
SINGLE_MQ:
while (true) {
try {
PullResult pullResult =
consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32); //遍历所有queue，挨个调用pull
System.out.printf("%s%n", pullResult);
putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
switch (pullResult.getPullStatus()) {
case FOUND:
break;
case NO_MATCHED_MSG:
break;
case NO_NEW_MSG:
break SINGLE_MQ;
case OFFSET_ILLEGAL:
break;
default:
break;
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
consumer.shutdown();
}
从上面的代码我们可以看出，pull和push用法上的基本差别就是：pull是客户端主动去拉取消息，push是注册了一个回调，当有新消息，该回调被调用。
但这还不是2者的最大区别，最大区别是：在pull里面，所有MessageQueue是向我们暴露的，我们需要自己去手动遍历所有的queue；而在push里面，我们只指定了订阅的topic，而MessageQueue是向我们隐藏的，在其内部做了“负载均衡”。
而上面的pull的代码，我们手动遍历了所有的queue，没有负载均衡！！！
如何实现Pull的负载均衡？？？
查看源码可以发现，
DefaultMQPullConsumer 给我们提供了负载均衡的接口：fetchMessageQueuesInBalance(String topic)
此时，我们可以自己实现循环去手动Pull并实现负载均衡
1.手动实现负载均衡
public class Test2 {
public static void main(String[] args) throws MQClientException, InterruptedException {
String namesrvAddr = "";
//消费组
DefaultMQPullConsumer pullConsumer = new DefaultMQPullConsumer("GroupName1");
//MQ NameService地址
pullConsumer.setNamesrvAddr(namesrvAddr);
//负载均衡模式
pullConsumer.setMessageModel(MessageModel.CLUSTERING);
//需要处理的消息topic
pullConsumer.start();
while (true) {
boolean waiting = true;
Set<MessageQueue> mqs = pullConsumer.fetchMessageQueuesInBalance("TopicTest1");
//未获取到负载均衡的时候，等待1S重新获取
if (!CollectionUtils.isEmpty(mqs)) {
waiting = false;
Thread.sleep(1000L);
}
for (MessageQueue mq : mqs) {
System.out.printf("Consume from the queue: " + mq + "%n");
SINGLE_MQ:
while (true) {
long offset = pullConsumer.fetchConsumeOffset(mq, false);
try {
PullResult pullResult =
pullConsumer.pullBlockIfNotFound(mq, null, offset, 32); //遍历所有queue，挨个调用pull
System.out.printf("%s%n", pullResult);
switch (pullResult.getPullStatus()) {
case FOUND:
offset = pullResult.getNextBeginOffset();
pullConsumer.updateConsumeOffset(mq, offset);
break;
case NO_MATCHED_MSG:
break;
case NO_NEW_MSG:
break SINGLE_MQ;
case OFFSET_ILLEGAL:
break;
default:
break;
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
if(waiting){
Thread.sleep(100L);
}
}
}
}
2.RocketMQ Pull模式下提供的负载均衡样例（基于MQPullConsumerScheduleService）
在MQPullConsumer这个类里面，有一个MessageQueueListener，它的目的就是当queue发生变化的时候，通知Consumer。也正是这个借口，帮助我们在Pull模式里面，实现负载均衡。
注意，这个接口在MQPushConsumer里面是没有的，那里面有的是上面代码里的MessageListener。
void registerMessageQueueListener(final String topic, final MessageQueueListener listener);
public interface MessageQueueListener {
void messageQueueChanged(final String topic, final Set<MessageQueue> mqAll,
final Set<MessageQueue> mqDivided);
}
有了这个Listener，我们就可以动态的知道当前的Consumer分摊到了几个MessageQueue。然后对这些MessageQueue，我们可以开个线程池来消费。
public static void main(String[] args) throws MQClientException {
//消费组
final MQPullConsumerScheduleService scheduleService = new MQPullConsumerScheduleService("GroupName1");
//MQ NameService地址
scheduleService.getDefaultMQPullConsumer().setNamesrvAddr(namesrvAddr);
//负载均衡模式
scheduleService.setMessageModel(MessageModel.CLUSTERING);
//需要处理的消息topic
scheduleService.registerPullTaskCallback("TopicTest1", new PullTaskCallback() {
@Override
public void doPullTask(MessageQueue mq, PullTaskContext context) {
MQPullConsumer consumer = context.getPullConsumer();
try {
long offset = consumer.fetchConsumeOffset(mq, false);
if (offset < 0)
offset = 0;
PullResult pullResult = consumer.pull(mq, "*", offset, 32);
System.out.printf("%s%n", offset + "\t" + mq + "\t" + pullResult);
switch (pullResult.getPullStatus()) {
case FOUND:
break;
case NO_MATCHED_MSG:
break;
case NO_NEW_MSG:
case OFFSET_ILLEGAL:
break;
default:
break;
}
consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
context.setPullNextDelayTimeMillis(100);
} catch (Exception e) {
e.printStackTrace();
}
}
});
scheduleService.start();
}
