本文整理汇总了Java中com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer类的典型用法代码示例。如果您正苦于以下问题：Java DefaultMQPullConsumer类的具体用法？Java DefaultMQPullConsumer怎么用？Java DefaultMQPullConsumer使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
DefaultMQPullConsumer类属于com.alibaba.rocketmq.client.consumer包，在下文中一共展示了DefaultMQPullConsumer类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。
示例1: main
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
consumer.start();
try {
MessageQueue mq = new MessageQueue();
mq.setQueueId(0);
mq.setTopic("TopicTest3");
mq.setBrokerName("vivedeMacBook-Pro.local");
long offset = 26;
long beginTime = System.currentTimeMillis();
PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, offset, 32);
System.out.println(System.currentTimeMillis() - beginTime);
System.out.println(pullResult);
}
catch (Exception e) {
e.printStackTrace();
}
consumer.shutdown();
}
开发者ID:y123456yz，项目名称:reading-and-annotate-rocketmq-3.4.6，代码行数:24，代码来源:PullConsumerTest.java
示例2: getServerAddr
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
private static String getServerAddr(Object mqClient) {
String addr = "mq:rocket://";
if (mqClient.getClass().getSimpleName().equals("DefaultMQProducer")) {
addr += ((DefaultMQProducer) mqClient).getNamesrvAddr();
}
else if (mqClient.getClass().getSimpleName().equals("DefaultMQPushConsumer")) {
addr += ((DefaultMQPushConsumer) mqClient).getNamesrvAddr();
}
else if (mqClient.getClass().getSimpleName().equals("DefaultMQPullConsumer")) {
addr += ((DefaultMQPullConsumer) mqClient).getNamesrvAddr();
}
return addr;
}
开发者ID:uavorg，项目名称:uavstack，代码行数:17，代码来源:RocketmqIT.java
示例3: main
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
consumer.start();
try {
MessageQueue mq = new MessageQueue();
mq.setQueueId(0);
mq.setTopic("TopicTest3");
mq.setBrokerName("vivedeMacBook-Pro.local");
long offset = 26;
long beginTime = System.currentTimeMillis();
PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, offset, 32);
System.out.println(System.currentTimeMillis() - beginTime);
System.out.println(pullResult);
}
catch (Exception e) {
e.printStackTrace();
}
consumer.shutdown();
}
开发者ID:medusar，项目名称:rocketmq-commet，代码行数:27，代码来源:PullConsumerTest.java
示例4: initEnv
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
@Before
public void initEnv() {
consumer = new DefaultMQPullConsumer("PushConsumer");
consumer.setNamesrvAddr("172.27.101.67:9876");
mq = new MessageQueue();
mq.setQueueId(0);
mq.setTopic("Topic3");
mq.setBrokerName("broker-a");
offsetStore = consumer.getDefaultMQPullConsumerImpl().getOffsetStore();
try {
consumer.start();
} catch (MQClientException e) {
e.printStackTrace();
}
}
开发者ID:bigdatafly，项目名称:flume-rocketmq-sink，代码行数:17，代码来源:PullConsumerTest.java
示例5: initConsumer
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
private DefaultMQPullConsumer initConsumer(String _instanceName) throws MQClientException,
InterruptedException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(consumerGroup);
if (StringUtils.isNotBlank(_instanceName)) {
consumer.setInstanceName(_instanceName);
}
else {
consumer.setInstanceName(instanceName);
}
MessageQueueListener listener = new MessageQueueListener() {
@Override
public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {
System.out.println("Topic=" + topic);
System.out.println("mqAll:" + mqAll);
System.out.println("mqDivided:" + mqDivided);
}
};
consumer.registerMessageQueueListener(topic, listener);
Thread.sleep(2000);
return consumer;
}
开发者ID:diwayou，项目名称:rocketmq-all-trans，代码行数:22，代码来源:PullConsumerTest.java
示例6: resetOffsetByTimestamp
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
/**
* Reset consumer topic offset according to time
*
* @param messageModel
* which model
* @param instanceName
* which instance
* @param consumerGroup
* consumer group
* @param topic
* topic
* @param timestamp
* time
* @throws Exception
*/
public static void resetOffsetByTimestamp(//
final MessageModel messageModel,//
final String instanceName,//
final String consumerGroup, //
final String topic, //
final long timestamp) throws Exception {
final Logger log = ClientLogger.getLog();
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(consumerGroup);
consumer.setInstanceName(instanceName);
consumer.setMessageModel(messageModel);
consumer.start();
Set<MessageQueue> mqs = null;
try {
mqs = consumer.fetchSubscribeMessageQueues(topic);
if (mqs != null && !mqs.isEmpty()) {
TreeSet<MessageQueue> mqsNew = new TreeSet<MessageQueue>(mqs);
for (MessageQueue mq : mqsNew) {
long offset = consumer.searchOffset(mq, timestamp);
if (offset >= 0) {
consumer.updateConsumeOffset(mq, offset);
log.info("resetOffsetByTimestamp updateConsumeOffset success, {} {} {}",
consumerGroup, offset, mq);
}
}
}
}
catch (Exception e) {
log.warn("resetOffsetByTimestamp Exception", e);
throw e;
}
finally {
if (mqs != null) {
consumer.getDefaultMQPullConsumerImpl().getOffsetStore().persistAll(mqs);
}
consumer.shutdown();
}
}
开发者ID:y123456yz，项目名称:reading-and-annotate-rocketmq-3.4.6，代码行数:55，代码来源:MQHelper.java
示例7: main
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
consumer.start();
Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest1");
for (MessageQueue mq : mqs) {
System.out.println("Consume from the queue: " + mq);
SINGLE_MQ: while (true) {
try {
PullResult pullResult =
consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
System.out.println(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();
}
开发者ID:y123456yz，项目名称:reading-and-annotate-rocketmq-3.4.6，代码行数:36，代码来源:PullConsumer.java
示例8: resetOffsetByTimestamp
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
/**
* 根据时间戳来重置一个订阅组的消费进度
*
* @param messageModel
* 广播消费还是集群消费
* @param instanceName
* 实例名称，保持与工作Consumer一致。
* @param consumerGroup
* 订阅组
* @param topic
* topic
* @param timestamp
* 时间戳
* @throws Exception
*/
public static void resetOffsetByTimestamp(//
final MessageModel messageModel,//
final String instanceName,//
final String consumerGroup, //
final String topic, //
final long timestamp) throws Exception {
final Logger log = ClientLogger.getLog();
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(consumerGroup);
consumer.setInstanceName(instanceName);
consumer.setMessageModel(messageModel);
consumer.start();
Set<MessageQueue> mqs = null;
try {
mqs = consumer.fetchSubscribeMessageQueues(topic);
if (mqs != null && !mqs.isEmpty()) {
TreeSet<MessageQueue> mqsNew = new TreeSet<MessageQueue>(mqs);
for (MessageQueue mq : mqsNew) {
long offset = consumer.searchOffset(mq, timestamp);
if (offset >= 0) {
consumer.updateConsumeOffset(mq, offset);
log.info("resetOffsetByTimestamp updateConsumeOffset success, {} {} {}",
consumerGroup, offset, mq);
}
}
}
}
catch (Exception e) {
log.warn("resetOffsetByTimestamp Exception", e);
throw e;
}
finally {
if (mqs != null) {
consumer.getDefaultMQPullConsumerImpl().getOffsetStore().persistAll(mqs);
}
consumer.shutdown();
}
}
开发者ID:diwayou，项目名称:rocketmq-all-trans，代码行数:55，代码来源:MQHelper.java
示例9: main
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
public static void main(String[] args) throws SocketException {
MixAll.printObjectProperties(null, new ClientConfig());
System.out.println("----------------------------------------------");
MixAll.printObjectProperties(null, new DefaultMQProducer());
System.out.println("----------------------------------------------");
MixAll.printObjectProperties(null, new TransactionMQProducer());
System.out.println("----------------------------------------------");
MixAll.printObjectProperties(null, new DefaultMQPushConsumer());
System.out.println("----------------------------------------------");
MixAll.printObjectProperties(null, new DefaultMQPullConsumer());
}
开发者ID:diwayou，项目名称:rocketmq-all-trans，代码行数:12，代码来源:ClientConfigTest.java
示例10: main
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
consumer.start();
Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest");
for (MessageQueue mq : mqs) {
System.out.println("Consume from the queue: " + mq);
SINGLE_MQ: while (true) {
try {
PullResult pullResult =
consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
System.out.println(pullResult);
putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
switch (pullResult.getPullStatus()) {
case FOUND:
// TODO
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();
}
开发者ID:diwayou，项目名称:rocketmq-all-trans，代码行数:37，代码来源:PullConsumer.java
示例11: main
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
public static void main(String[] args) throws MQClientException {
final MQPullConsumerScheduleService scheduleService = new MQPullConsumerScheduleService("GroupName1");
scheduleService.setMessageModel(MessageModel.CLUSTERING);
scheduleService.registerPullTaskCallback("TopicTest1", new PullTaskCallback() {
@Override
public void doPullTask(MessageQueue mq, PullTaskContext context) {
DefaultMQPullConsumer consumer = scheduleService.getDefaultMQPullConsumer();
try {
long offset = consumer.fetchConsumeOffset(mq, false);
if (offset < 0)
offset = 0;
PullResult pullResult = consumer.pull(mq, "*", offset, 32);
System.out.println(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());
}
catch (Exception e) {
e.printStackTrace();
}
}
});
scheduleService.start();
}
开发者ID:diwayou，项目名称:rocketmq-all-trans，代码行数:39，代码来源:PullScheduleService.java
示例12: PullMsgFromMqTask
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
public PullMsgFromMqTask(DefaultMQPullConsumer consumer, MessageQueue partition, AtomicBoolean isOK,
CountDownLatch stopLatch) {
this.consumer = consumer;
this.partition = partition;
this.isOK = isOK;
this.stopLatch = stopLatch;
}
开发者ID:brucechan0921，项目名称:RocketMQ-3.0.8，代码行数:8，代码来源:PullMsgFromMqTask.java
示例13: getMessageQueueOffset
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
private static long getMessageQueueOffset(DefaultMQPullConsumer consumer, MessageQueue mq, boolean fromStore)
throws MQClientException {
long offset = consumer.fetchConsumeOffset(mq, fromStore);
offset = (offset < 0 ? 0 : offset);
System.out.println("offset:" + offset);
return offset;
}
开发者ID:brucechan0921，项目名称:RocketMQ-3.0.8，代码行数:8，代码来源:PullMsgFromMqTask.java
示例14: DefaultMQPullConsumerImpl
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
public DefaultMQPullConsumerImpl(final DefaultMQPullConsumer defaultMQPullConsumer, final RPCHook rpcHook) {
this.defaultMQPullConsumer = defaultMQPullConsumer;
this.rpcHook = rpcHook;
}
开发者ID:y123456yz，项目名称:reading-and-annotate-rocketmq-3.4.6，代码行数:5，代码来源:DefaultMQPullConsumerImpl.java
示例15: getDefaultMQPullConsumer
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
public DefaultMQPullConsumer getDefaultMQPullConsumer() {
return defaultMQPullConsumer;
}
开发者ID:y123456yz，项目名称:reading-and-annotate-rocketmq-3.4.6，代码行数:4，代码来源:DefaultMQPullConsumerImpl.java
示例16: execute
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
@Override
public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer(MixAll.TOOLS_CONSUMER_GROUP);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
defaultMQPullConsumer.setInstanceName(Long.toString(System.currentTimeMillis()));
try {
//解析 sh mqadmin queryMsgByOffset 命令携带的各种参数信息
String topic = commandLine.getOptionValue('t').trim();
String brokerName = commandLine.getOptionValue('b').trim();
String queueId = commandLine.getOptionValue('i').trim();
String offset = commandLine.getOptionValue('o').trim();
MessageQueue mq = new MessageQueue();
mq.setTopic(topic);
mq.setBrokerName(brokerName);
mq.setQueueId(Integer.parseInt(queueId));
defaultMQPullConsumer.start();
defaultMQAdminExt.start();
PullResult pullResult = defaultMQPullConsumer.pull(mq, "*", Long.parseLong(offset), 1);
if (pullResult != null) {
switch (pullResult.getPullStatus()) {
case FOUND:
QueryMsgByIdSubCommand.queryById(defaultMQAdminExt, pullResult.getMsgFoundList().get(0)
.getMsgId());
break;
case NO_MATCHED_MSG:
case NO_NEW_MSG:
case OFFSET_ILLEGAL:
default:
break;
}
}
}
catch (Exception e) {
e.printStackTrace();
}
finally {
defaultMQPullConsumer.shutdown();
defaultMQAdminExt.shutdown();
}
}
开发者ID:y123456yz，项目名称:reading-and-annotate-rocketmq-3.4.6，代码行数:47，代码来源:QueryMsgByOffsetSubCommand.java
示例17: MqPullConsumer
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
public MqPullConsumer(String namesrvAddr){
consumer = new DefaultMQPullConsumer();
}
开发者ID:TFdream，项目名称:mq-in-action，代码行数:4，代码来源:MqPullConsumer.java
示例18: execute
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
@Override
public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer(MixAll.TOOLS_CONSUMER_GROUP);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
defaultMQPullConsumer.setInstanceName(Long.toString(System.currentTimeMillis()));
try {
String topic = commandLine.getOptionValue('t').trim();
String brokerName = commandLine.getOptionValue('b').trim();
String queueId = commandLine.getOptionValue('i').trim();
String offset = commandLine.getOptionValue('o').trim();
MessageQueue mq = new MessageQueue();
mq.setTopic(topic);
mq.setBrokerName(brokerName);
mq.setQueueId(Integer.parseInt(queueId));
defaultMQPullConsumer.start();
defaultMQAdminExt.start();
PullResult pullResult = defaultMQPullConsumer.pull(mq, "*", Long.parseLong(offset), 1);
if (pullResult != null) {
switch (pullResult.getPullStatus()) {
case FOUND:
QueryMsgByIdSubCommand.queryById(defaultMQAdminExt, pullResult.getMsgFoundList().get(0)
.getMsgId());
break;
case NO_MATCHED_MSG:
case NO_NEW_MSG:
case OFFSET_ILLEGAL:
default:
break;
}
}
}
catch (Exception e) {
e.printStackTrace();
}
finally {
defaultMQPullConsumer.shutdown();
defaultMQAdminExt.shutdown();
}
}
开发者ID:medusar，项目名称:rocketmq-commet，代码行数:46，代码来源:QueryMsgByOffsetSubCommand.java
示例19: queryMsgByOffset
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
@CmdTrace(cmdClazz = QueryMsgByOffsetSubCommand.class)
public Table queryMsgByOffset(String topicName, String brokerName, String queueId, String offset)
throws Throwable {
Throwable t = null;
DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer(MixAll.TOOLS_CONSUMER_GROUP);
defaultMQPullConsumer.setInstanceName(Long.toString(System.currentTimeMillis()));
try {
MessageQueue mq = new MessageQueue();
mq.setTopic(topicName);
mq.setBrokerName(brokerName);
mq.setQueueId(Integer.parseInt(queueId));
defaultMQPullConsumer.start();
PullResult pullResult = defaultMQPullConsumer.pull(mq, "*", Long.parseLong(offset), 1);
if (pullResult != null) {
switch (pullResult.getPullStatus()) {
case FOUND:
Table table = queryMsgById(pullResult.getMsgFoundList().get(0).getMsgId());
return table;
case NO_MATCHED_MSG:
case NO_NEW_MSG:
case OFFSET_ILLEGAL:
default:
break;
}
}
else {
throw new IllegalStateException("pullResult is null");
}
}
catch (Throwable e) {
logger.error(e.getMessage(), e);
t = e;
}
finally {
defaultMQPullConsumer.shutdown();
}
throw t;
}
开发者ID:LXLun，项目名称:RocketMQMonitor，代码行数:43，代码来源:MessageService.java
示例20: DefaultMQPullConsumerImpl
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer; //导入依赖的package包/类
public DefaultMQPullConsumerImpl(final DefaultMQPullConsumer defaultMQPullConsumer) {
this.defaultMQPullConsumer = defaultMQPullConsumer;
}
开发者ID:diwayou，项目名称:rocketmq-all-trans，代码行数:4，代码来源:DefaultMQPullConsumerImpl.java
注：本文中的com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论