package simple;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import java.util.HashSet;
import java.util.Set;
/**
* 拉模式
* Created by BaiLi
*/
public class PullConsumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer pullConsumer = new DefaultMQPullConsumer("SimplePullConsumer");
pullConsumer.setNamesrvAddr("192.168.43.137:9876");//执行nameserver地址
Set<String> topics = new HashSet<>();
topics.add("simple");//添加Topic
topics.add("TopicTest");
pullConsumer.setRegisterTopics(topics);
pullConsumer.start();
while (true) { //循环拉取消息
pullConsumer.getRegisterTopics().forEach(n -> {
try {
Set<MessageQueue> messageQueues = pullConsumer.fetchSubscribeMessageQueues(n);//获取主题中的Queue
messageQueues.forEach(l -> {
try {
//获取Queue中的偏移量
long offset = pullConsumer.getOffsetStore().readOffset(l, ReadOffsetType.READ_FROM_MEMORY);
if (offset < 0) {
offset = pullConsumer.getOffsetStore().readOffset(l, ReadOffsetType.READ_FROM_STORE);
}
if (offset < 0) {
offset = pullConsumer.maxOffset(l);
}
if (offset < 0) {
offset = 0;
}
//拉取Queue中的消息。每次获取32条
PullResult pullResult = pullConsumer.pull(l, "*", offset, 32);
System.out.printf("循环拉取消息ing %s%n",pullResult);
switch (pullResult.getPullStatus()) {
case FOUND:
pullResult.getMsgFoundList().forEach(p -> {
System.out.printf("拉取消息成功%s%n", p);
});
//更新偏移量
pullConsumer.updateConsumeOffset(l, pullResult.getNextBeginOffset());
}
} catch (MQClientException e) {
e.printStackTrace();
} catch (RemotingException e) {
e.printStackTrace();
} catch (MQBrokerException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
} catch (MQClientException e) {
e.printStackTrace();
}
});
}
}
}
Rocket_API_BaiLi_笔记版.zip （61个子文件） 
Rocket_API_BaiLi 
pom.xml 802B src test java main resources java schedule ScheduleConsumer.java 1KB TimeConsumer.java 1KB ScheduleProducer.java 1KB TimeProducer.java 2KB order OrderConsumer.java 1KB OrderProducer.java 2KB simple OnewayProducer.java 1002B AsyncProducer.java 2KB PullConsumer.java 3KB PullLiteConsumerAssign.java 1KB Consumer.java 1KB PullLiteConsumer.java 797B SyncProducer.java 1KB filter TagFilterProducer.java 1KB SqlFilterConsumer.java 2KB TagFilterConsumer.java 1KB SqlFilterProducer.java 1KB transaction TransactionListenerImpl.java 2KB TransactionProducer.java 2KB broadcast BroadcastConsumer.java 2KB batch BatchProducer.java 1KB SplitBatchProducer.java 3KB .idea uiDesigner.xml 9KB workspace.xml 7KB misc.xml 526B inspectionProfiles Project_Default.xml 368B target classes schedule TimeConsumer$1.class 2KB ScheduleProducer.class 2KB TimeConsumer.class 1KB TimeProducer.class 2KB ScheduleConsumer$1.class 2KB ScheduleConsumer.class 1KB order OrderConsumer$1.class 3KB OrderProducer$1.class 1KB OrderProducer.class 2KB OrderConsumer.class 1KB simple OnewayProducer.class 2KB Consumer.class 1KB AsyncProducer.class 2KB Consumer$1.class 2KB PullConsumer$1.class 707B PullLiteConsumerAssign.class 2KB PullLiteConsumer.class 1KB PullConsumer.class 5KB AsyncProducer$1.class 1KB SyncProducer.class 2KB filter SqlFilterConsumer$1.class 2KB TagFilterConsumer.class 1KB TagFilterProducer.class 2KB SqlFilterProducer.class 2KB TagFilterConsumer$1.class 2KB SqlFilterConsumer.class 2KB transaction TransactionProducer$1.class 841B TransactionListenerImpl.class 2KB TransactionProducer.class 3KB broadcast BroadcastConsumer$1.class 3KB BroadcastConsumer.class 2KB batch SplitBatchProducer.class 2KB ListSplitter.class 2KB BatchProducer.class 2KB资源评论
