本文整理汇总了Java中org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext类的典型用法代码示例。如果您正苦于以下问题:Java ConsumeOrderlyContext类的具体用法?Java ConsumeOrderlyContext怎么用?Java ConsumeOrderlyContext使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
ConsumeOrderlyContext类属于org.apache.rocketmq.client.consumer.listener包,在下文中一共展示了ConsumeOrderlyContext类的12个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Java代码示例。
示例1: testPullMessage_SuccessWithOrderlyService
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
@Test
public void testPullMessage_SuccessWithOrderlyService() throws Exception {
final CountDownLatch countDownLatch = new CountDownLatch(1);
final MessageExt[] messageExts = new MessageExt[1];
MessageListenerOrderly listenerOrderly = new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
messageExts[0] = msgs.get(0);
countDownLatch.countDown();
return null;
}
};
pushConsumer.registerMessageListener(listenerOrderly);
pushConsumer.getDefaultMQPushConsumerImpl().setConsumeMessageService(new ConsumeMessageOrderlyService(pushConsumer.getDefaultMQPushConsumerImpl(), listenerOrderly));
pushConsumer.getDefaultMQPushConsumerImpl().setConsumeOrderly(true);
pushConsumer.getDefaultMQPushConsumerImpl().doRebalance();
PullMessageService pullMessageService = mQClientFactory.getPullMessageService();
pullMessageService.executePullRequestLater(createPullRequest(), 100);
countDownLatch.await(10, TimeUnit.SECONDS);
assertThat(messageExts[0].getTopic()).isEqualTo(topic);
assertThat(messageExts[0].getBody()).isEqualTo(new byte[] {'a'});
}
开发者ID:lirenzuo,项目名称:rocketmq-rocketmq-all-4.1.0-incubating,代码行数:25,代码来源:DefaultMQPushConsumerTest.java
示例2: consumeMessage
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeOrderlyContext context) {
for (MessageExt msg : msgs) {
if (isDebug) {
if (listnerName != null && listnerName != "") {
logger.info(listnerName + ": " + msg);
} else {
logger.info(msg);
}
}
putMsg(msg);
msgBodys.addData(new String(msg.getBody()));
originMsgs.addData(msg);
}
return ConsumeOrderlyStatus.SUCCESS;
}
开发者ID:lirenzuo,项目名称:rocketmq-rocketmq-all-4.1.0-incubating,代码行数:19,代码来源:RMQOrderListener.java
示例3: main
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_message_consumer_group");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("TopicTest", "TagA || TagC || TagD");
// 顺序消费特点:不会有两个消费者共同消费任一队列,且当消费者数量小于队列数时,消费者会消费多个队列
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
//context.setAutoCommit(false);
System.out.printf(System.currentTimeMillis() + "," + Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
System.out.printf("Consumer Started.%n");
}
开发者ID:javahongxi,项目名称:whatsmars,代码行数:23,代码来源:Consumer.java
示例4: consumeMessage
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeOrderlyContext context) {
for (MessageExt msg : msgs) {
if (isDebug) {
if (listenerName != null && listenerName != "") {
logger.info(listenerName + ": " + msg);
} else {
logger.info(msg);
}
}
putMsg(msg);
msgBodys.addData(new String(msg.getBody()));
originMsgs.addData(msg);
}
return ConsumeOrderlyStatus.SUCCESS;
}
开发者ID:apache,项目名称:rocketmq,代码行数:19,代码来源:RMQOrderListener.java
示例5: receiveOrderMessage
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
public void receiveOrderMessage(){
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer(RocketMQConfiguration.ROCKETMQ_GROUP);
consumer.setNamesrvAddr(RocketMQConfiguration.ROCKETMQ_NAMESRV);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
try {
consumer.subscribe(RocketMQConfiguration.ROCKETMQ_TOPIC,"*");
//监听消息接受
consumer.registerMessageListener(new MessageListenerOrderly() {
AtomicLong consumeTimes = new AtomicLong(0);
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext consumeOrderlyContext) {
consumeOrderlyContext.setAutoCommit(false);
System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + new String(msgs.get(0).getBody()) + "%n");
System.out.println(consumeTimes.incrementAndGet());
if ((this.consumeTimes.get() % 2) == 0) {
return ConsumeOrderlyStatus.SUCCESS;
} else if ((this.consumeTimes.get() % 3) == 0) {
return ConsumeOrderlyStatus.ROLLBACK;
} else if ((this.consumeTimes.get() % 4) == 0) {
return ConsumeOrderlyStatus.COMMIT;
} else if ((this.consumeTimes.get() % 5) == 0) {
consumeOrderlyContext.setSuspendCurrentQueueTimeMillis(3000);
return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
} catch (MQClientException e) {
e.printStackTrace();
}
}
开发者ID:mumudemo,项目名称:mumu-rocketmq,代码行数:33,代码来源:RocketMQOrderConsumer.java
示例6: main
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("TopicTest", "TagA || TagC || TagD");
consumer.registerMessageListener(new MessageListenerOrderly() {
AtomicLong consumeTimes = new AtomicLong(0);
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
context.setAutoCommit(false);
System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
this.consumeTimes.incrementAndGet();
if ((this.consumeTimes.get() % 2) == 0) {
return ConsumeOrderlyStatus.SUCCESS;
} else if ((this.consumeTimes.get() % 3) == 0) {
return ConsumeOrderlyStatus.ROLLBACK;
} else if ((this.consumeTimes.get() % 4) == 0) {
return ConsumeOrderlyStatus.COMMIT;
} else if ((this.consumeTimes.get() % 5) == 0) {
context.setSuspendCurrentQueueTimeMillis(3000);
return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
System.out.printf("Consumer Started.%n");
}
开发者ID:lirenzuo,项目名称:rocketmq-rocketmq-all-4.1.0-incubating,代码行数:34,代码来源:Consumer.java
示例7: dealMessage
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
/**
* 原生dealMessage方法,可以重写此方法自定义序列化和返回消费成功的相关逻辑
*
* @param list 消息列表
* @param consumeOrderlyContext 上下文
* @return 处理结果
*/
public ConsumeOrderlyStatus dealMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
for(MessageExt messageExt : list) {
log.info("receive msgId: {}, tags : {}" , messageExt.getMsgId(), messageExt.getTags());
T t = parseMessage(messageExt);
Map<String, Object> ext = parseExtParam(messageExt);
if( null != t && !process(t, ext)) {
log.warn("consume fail , ask for re-consume , msgId: {}", messageExt.getMsgId());
return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
}
}
return ConsumeOrderlyStatus.SUCCESS;
}
开发者ID:maihaoche,项目名称:rocketmq-spring-boot-starter,代码行数:20,代码来源:AbstractMQPushConsumer.java
示例8: main
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("TopicTest", "TagA || TagC || TagD");
consumer.registerMessageListener(new MessageListenerOrderly() {
AtomicLong consumeTimes = new AtomicLong(0);
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
context.setAutoCommit(false);
System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
this.consumeTimes.incrementAndGet();
if ((this.consumeTimes.get() % 2) == 0) {
return ConsumeOrderlyStatus.SUCCESS;
} else if ((this.consumeTimes.get() % 3) == 0) {
return ConsumeOrderlyStatus.ROLLBACK;
} else if ((this.consumeTimes.get() % 4) == 0) {
return ConsumeOrderlyStatus.COMMIT;
} else if ((this.consumeTimes.get() % 5) == 0) {
context.setSuspendCurrentQueueTimeMillis(3000);
return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
System.out.printf("Consumer Started.%n");
}
开发者ID:apache,项目名称:rocketmq,代码行数:34,代码来源:Consumer.java
示例9: consumeMessageDirectly
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName) {
ConsumeMessageDirectlyResult result = new ConsumeMessageDirectlyResult();
result.setOrder(true);
List<MessageExt> msgs = new ArrayList<MessageExt>();
msgs.add(msg);
MessageQueue mq = new MessageQueue();
mq.setBrokerName(brokerName);
mq.setTopic(msg.getTopic());
mq.setQueueId(msg.getQueueId());
ConsumeOrderlyContext context = new ConsumeOrderlyContext(mq);
final long beginTime = System.currentTimeMillis();
log.info("consumeMessageDirectly receive new messge: {}", msg);
try {
ConsumeOrderlyStatus status = this.messageListener.consumeMessage(msgs, context);
if (status != null) {
switch (status) {
case COMMIT:
result.setConsumeResult(CMResult.CR_COMMIT);
break;
case ROLLBACK:
result.setConsumeResult(CMResult.CR_ROLLBACK);
break;
case SUCCESS:
result.setConsumeResult(CMResult.CR_SUCCESS);
break;
case SUSPEND_CURRENT_QUEUE_A_MOMENT:
result.setConsumeResult(CMResult.CR_LATER);
break;
default:
break;
}
} else {
result.setConsumeResult(CMResult.CR_RETURN_NULL);
}
} catch (Throwable e) {
result.setConsumeResult(CMResult.CR_THROW_EXCEPTION);
result.setRemark(RemotingHelper.exceptionSimpleDesc(e));
log.warn(String.format("consumeMessageDirectly exception: %s Group: %s Msgs: %s MQ: %s", //
RemotingHelper.exceptionSimpleDesc(e), //
ConsumeMessageOrderlyService.this.consumerGroup, //
msgs, //
mq), e);
}
result.setAutoCommit(context.isAutoCommit());
result.setSpentTimeMills(System.currentTimeMillis() - beginTime);
log.info("consumeMessageDirectly Result: {}", result);
return result;
}
开发者ID:lirenzuo,项目名称:rocketmq-rocketmq-all-4.1.0-incubating,代码行数:59,代码来源:ConsumeMessageOrderlyService.java
示例10: processConsumeResult
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
public boolean processConsumeResult(//
final List<MessageExt> msgs, //
final ConsumeOrderlyStatus status, //
final ConsumeOrderlyContext context, //
final ConsumeRequest consumeRequest//
) {
boolean continueConsume = true;
long commitOffset = -1L;
if (context.isAutoCommit()) {
switch (status) {
case COMMIT:
case ROLLBACK:
log.warn("the message queue consume result is illegal, we think you want to ack these message {}",
consumeRequest.getMessageQueue());
case SUCCESS:
commitOffset = consumeRequest.getProcessQueue().commit();
this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
break;
case SUSPEND_CURRENT_QUEUE_A_MOMENT:
this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
if (checkReconsumeTimes(msgs)) {
consumeRequest.getProcessQueue().makeMessageToCosumeAgain(msgs);
this.submitConsumeRequestLater(//
consumeRequest.getProcessQueue(), //
consumeRequest.getMessageQueue(), //
context.getSuspendCurrentQueueTimeMillis());
continueConsume = false;
} else {
commitOffset = consumeRequest.getProcessQueue().commit();
}
break;
default:
break;
}
} else {
switch (status) {
case SUCCESS:
this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
break;
case COMMIT:
commitOffset = consumeRequest.getProcessQueue().commit();
break;
case ROLLBACK:
consumeRequest.getProcessQueue().rollback();
this.submitConsumeRequestLater(//
consumeRequest.getProcessQueue(), //
consumeRequest.getMessageQueue(), //
context.getSuspendCurrentQueueTimeMillis());
continueConsume = false;
break;
case SUSPEND_CURRENT_QUEUE_A_MOMENT:
this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
if (checkReconsumeTimes(msgs)) {
consumeRequest.getProcessQueue().makeMessageToCosumeAgain(msgs);
this.submitConsumeRequestLater(//
consumeRequest.getProcessQueue(), //
consumeRequest.getMessageQueue(), //
context.getSuspendCurrentQueueTimeMillis());
continueConsume = false;
}
break;
default:
break;
}
}
if (commitOffset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), commitOffset, false);
}
return continueConsume;
}
开发者ID:lirenzuo,项目名称:rocketmq-rocketmq-all-4.1.0-incubating,代码行数:73,代码来源:ConsumeMessageOrderlyService.java
示例11: consumeMessageDirectly
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, String brokerName) {
ConsumeMessageDirectlyResult result = new ConsumeMessageDirectlyResult();
result.setOrder(true);
List<MessageExt> msgs = new ArrayList<MessageExt>();
msgs.add(msg);
MessageQueue mq = new MessageQueue();
mq.setBrokerName(brokerName);
mq.setTopic(msg.getTopic());
mq.setQueueId(msg.getQueueId());
ConsumeOrderlyContext context = new ConsumeOrderlyContext(mq);
final long beginTime = System.currentTimeMillis();
log.info("consumeMessageDirectly receive new message: {}", msg);
try {
ConsumeOrderlyStatus status = this.messageListener.consumeMessage(msgs, context);
if (status != null) {
switch (status) {
case COMMIT:
result.setConsumeResult(CMResult.CR_COMMIT);
break;
case ROLLBACK:
result.setConsumeResult(CMResult.CR_ROLLBACK);
break;
case SUCCESS:
result.setConsumeResult(CMResult.CR_SUCCESS);
break;
case SUSPEND_CURRENT_QUEUE_A_MOMENT:
result.setConsumeResult(CMResult.CR_LATER);
break;
default:
break;
}
} else {
result.setConsumeResult(CMResult.CR_RETURN_NULL);
}
} catch (Throwable e) {
result.setConsumeResult(CMResult.CR_THROW_EXCEPTION);
result.setRemark(RemotingHelper.exceptionSimpleDesc(e));
log.warn(String.format("consumeMessageDirectly exception: %s Group: %s Msgs: %s MQ: %s",
RemotingHelper.exceptionSimpleDesc(e),
ConsumeMessageOrderlyService.this.consumerGroup,
msgs,
mq), e);
}
result.setAutoCommit(context.isAutoCommit());
result.setSpentTimeMills(System.currentTimeMillis() - beginTime);
log.info("consumeMessageDirectly Result: {}", result);
return result;
}
开发者ID:apache,项目名称:rocketmq,代码行数:59,代码来源:ConsumeMessageOrderlyService.java
示例12: processConsumeResult
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; //导入依赖的package包/类
public boolean processConsumeResult(
final List<MessageExt> msgs,
final ConsumeOrderlyStatus status,
final ConsumeOrderlyContext context,
final ConsumeRequest consumeRequest
) {
boolean continueConsume = true;
long commitOffset = -1L;
if (context.isAutoCommit()) {
switch (status) {
case COMMIT:
case ROLLBACK:
log.warn("the message queue consume result is illegal, we think you want to ack these message {}",
consumeRequest.getMessageQueue());
case SUCCESS:
commitOffset = consumeRequest.getProcessQueue().commit();
this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
break;
case SUSPEND_CURRENT_QUEUE_A_MOMENT:
this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
if (checkReconsumeTimes(msgs)) {
consumeRequest.getProcessQueue().makeMessageToCosumeAgain(msgs);
this.submitConsumeRequestLater(
consumeRequest.getProcessQueue(),
consumeRequest.getMessageQueue(),
context.getSuspendCurrentQueueTimeMillis());
continueConsume = false;
} else {
commitOffset = consumeRequest.getProcessQueue().commit();
}
break;
default:
break;
}
} else {
switch (status) {
case SUCCESS:
this.getConsumerStatsManager().incConsumeOKTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
break;
case COMMIT:
commitOffset = consumeRequest.getProcessQueue().commit();
break;
case ROLLBACK:
consumeRequest.getProcessQueue().rollback();
this.submitConsumeRequestLater(
consumeRequest.getProcessQueue(),
consumeRequest.getMessageQueue(),
context.getSuspendCurrentQueueTimeMillis());
continueConsume = false;
break;
case SUSPEND_CURRENT_QUEUE_A_MOMENT:
this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
if (checkReconsumeTimes(msgs)) {
consumeRequest.getProcessQueue().makeMessageToCosumeAgain(msgs);
this.submitConsumeRequestLater(
consumeRequest.getProcessQueue(),
consumeRequest.getMessageQueue(),
context.getSuspendCurrentQueueTimeMillis());
continueConsume = false;
}
break;
default:
break;
}
}
if (commitOffset >= 0 && !consumeRequest.getProcessQueue().isDropped()) {
this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), commitOffset, false);
}
return continueConsume;
}
开发者ID:apache,项目名称:rocketmq,代码行数:73,代码来源:ConsumeMessageOrderlyService.java
注:本文中的org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext类示例整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论