Java ConsumeConcurrentlyStatus类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Java›Java编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Java中org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus类的典型用法代码示例。如果您正苦于以下问题：Java ConsumeConcurrentlyStatus类的具体用法？Java ConsumeConcurrentlyStatus怎么用？Java ConsumeConcurrentlyStatus使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

ConsumeConcurrentlyStatus类属于org.apache.rocketmq.client.consumer.listener包，在下文中一共展示了ConsumeConcurrentlyStatus类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。

示例1: receiveBroadcastMessage

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public void receiveBroadcastMessage(){
    DefaultMQPushConsumer consumer=new DefaultMQPushConsumer(RocketMQConfiguration.ROCKETMQ_GROUP);
    consumer.setNamesrvAddr(RocketMQConfiguration.ROCKETMQ_NAMESRV);
    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    consumer.setMessageModel(MessageModel.BROADCASTING);
    try {
        consumer.subscribe(RocketMQConfiguration.ROCKETMQ_TOPIC,"*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    } catch (MQClientException e) {
        e.printStackTrace();
    }
}

开发者ID:mumudemo，项目名称:mumu-rocketmq，代码行数:20，代码来源:RocketMQBroadcastConsumer.java

示例2: receiveTransactionMessage

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
/**
 * 接收消息
 */
public void receiveTransactionMessage() {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMQConfiguration.ROCKETMQ_GROUP);
    consumer.setNamesrvAddr(RocketMQConfiguration.ROCKETMQ_NAMESRV);
    consumer.setVipChannelEnabled(false);
    consumer.setConsumeThreadMax(20);
    consumer.setConsumeThreadMin(10);
    try {
        consumer.subscribe(RocketMQConfiguration.ROCKETMQ_TOPIC, "transaction");
        //程序第一次启动从消息队列头取数据
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(
                new MessageListenerConcurrently() {
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext Context) {
                        Message msg = list.get(0);
                        System.out.println(msg.toString());
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                }
        );
        consumer.start();
    } catch (MQClientException e) {
        e.printStackTrace();
    }
}

开发者ID:mumudemo，项目名称:mumu-rocketmq，代码行数:28，代码来源:RocketMQTransactionConsumer.java

示例3: receiveMessage

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
/**
 * 接收消息
 */
public void receiveMessage(){
    DefaultMQPushConsumer consumer =new DefaultMQPushConsumer(RocketMQConfiguration.ROCKETMQ_GROUP);
    consumer.setNamesrvAddr(RocketMQConfiguration.ROCKETMQ_NAMESRV);
    consumer.setVipChannelEnabled(false);
    try {
        consumer.subscribe(RocketMQConfiguration.ROCKETMQ_TOPIC, "schedule");
        //程序第一次启动从消息队列头取数据
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(
                new MessageListenerConcurrently() {
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,ConsumeConcurrentlyContext Context) {
                        Message msg = list.get(0);
                        System.out.println(msg.toString());
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                }
        );
        consumer.start();
    } catch (MQClientException e) {
        e.printStackTrace();
    }
}

开发者ID:mumudemo，项目名称:mumu-rocketmq，代码行数:26，代码来源:RocketMQScheduleConsumer.java

示例4: receiveMessage

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
/**
 * 接收消息
 */
public void receiveMessage(){
    DefaultMQPushConsumer consumer =new DefaultMQPushConsumer(RocketMQConfiguration.ROCKETMQ_GROUP);
    consumer.setNamesrvAddr(RocketMQConfiguration.ROCKETMQ_NAMESRV);
    consumer.setVipChannelEnabled(false);
    try {
        consumer.subscribe(RocketMQConfiguration.ROCKETMQ_TOPIC, "bactch");
        //程序第一次启动从消息队列头取数据
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(
                new MessageListenerConcurrently() {
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,ConsumeConcurrentlyContext Context) {
                        Message msg = list.get(0);
                        System.out.println(msg.toString());
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                }
        );
        consumer.start();
    } catch (MQClientException e) {
        e.printStackTrace();
    }
}

开发者ID:mumudemo，项目名称:mumu-rocketmq，代码行数:26，代码来源:RocketMQBatchConsumer.java

示例5: receiveFilterMessage

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
/**
 * 接收消息
 */
public void receiveFilterMessage(){
    DefaultMQPushConsumer consumer =new DefaultMQPushConsumer(RocketMQConfiguration.ROCKETMQ_GROUP);
    consumer.setNamesrvAddr(RocketMQConfiguration.ROCKETMQ_NAMESRV);
    consumer.setVipChannelEnabled(false);
    try {
        consumer.subscribe(RocketMQConfiguration.ROCKETMQ_TOPIC, MessageSelector.bySql("a between 5 and 8"));
        //程序第一次启动从消息队列头取数据
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(
                new MessageListenerConcurrently() {
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext Context) {
                        Message msg = list.get(0);
                        System.out.println(msg.toString());
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                }
        );
        consumer.start();
    } catch (MQClientException e) {
        e.printStackTrace();
    }
}

开发者ID:mumudemo，项目名称:mumu-rocketmq，代码行数:26，代码来源:RocketMQFilterConsumer.java

示例6: main

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException, MQClientException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_1");

    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

    consumer.setMessageModel(MessageModel.BROADCASTING);

    consumer.subscribe("TopicTest", "TagA || TagC || TagD");

    consumer.registerMessageListener(new MessageListenerConcurrently() {

        @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("Broadcast Consumer Started.%n");
}

开发者ID:lirenzuo，项目名称:rocketmq-rocketmq-all-4.1.0-incubating，代码行数:23，代码来源:PushConsumer.java

示例7: main

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException, MQClientException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");
    consumer.subscribe("Jodie_topic_1023", "*");
    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    //wrong time format 2017_0422_221800
    consumer.setConsumeTimestamp("20170422221800");
    consumer.registerMessageListener(new MessageListenerConcurrently() {

        /**

         */
        @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");
}

开发者ID:lirenzuo，项目名称:rocketmq-rocketmq-all-4.1.0-incubating，代码行数:21，代码来源:PushConsumer.java

示例8: main

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException, MQClientException, IOException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupNamecc4");

    ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
    File classFile = new File(classLoader.getResource("MessageFilterImpl.java").getFile());

    String filterCode = MixAll.file2String(classFile);
    consumer.subscribe("TopicTest", "org.apache.rocketmq.example.filter.MessageFilterImpl",
        filterCode);

    consumer.registerMessageListener(new MessageListenerConcurrently() {

        @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");
}

开发者ID:lirenzuo，项目名称:rocketmq-rocketmq-all-4.1.0-incubating，代码行数:25，代码来源:Consumer.java

示例9: testPullMessage_Success

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
@Test
public void testPullMessage_Success() throws InterruptedException, RemotingException, MQBrokerException {
    final CountDownLatch countDownLatch = new CountDownLatch(1);
    final MessageExt[] messageExts = new MessageExt[1];
    pushConsumer.getDefaultMQPushConsumerImpl().setConsumeMessageService(new ConsumeMessageConcurrentlyService(pushConsumer.getDefaultMQPushConsumerImpl(), new MessageListenerConcurrently() {
        @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            messageExts[0] = msgs.get(0);
            countDownLatch.countDown();
            return null;
        }
    }));

    PullMessageService pullMessageService = mQClientFactory.getPullMessageService();
    pullMessageService.executePullRequestImmediately(createPullRequest());
    countDownLatch.await();
    assertThat(messageExts[0].getTopic()).isEqualTo(topic);
    assertThat(messageExts[0].getBody()).isEqualTo(new byte[] {'a'});
}

开发者ID:lirenzuo，项目名称:rocketmq-rocketmq-all-4.1.0-incubating，代码行数:20，代码来源:DefaultMQPushConsumerTest.java

示例10: consumeMessage

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
    for (MessageExt msg : msgs) {
        msgIndex.getAndIncrement();
        if (isDebug) {
            if (listnerName != null && listnerName != "") {
                logger.info(listnerName + ":" + msgIndex.get() + ":"
                    + String.format("msgid:%s broker:%s queueId:%s offset:%s",
                    msg.getMsgId(), msg.getStoreHost(), msg.getQueueId(),
                    msg.getQueueOffset()));
            } else {
                logger.info(msg);
            }
        }

        msgBodys.addData(new String(msg.getBody()));
        originMsgs.addData(msg);
        if (originMsgIndex != null) {
            originMsgIndex.put(new String(msg.getBody()), msg);
        }
    }
    return consumeStatus;
}

开发者ID:lirenzuo，项目名称:rocketmq-rocketmq-all-4.1.0-incubating，代码行数:24，代码来源:RMQNormalListner.java

示例11: testStartTwoConsumerAndOneConsumerFail

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
@Test
public void testStartTwoConsumerAndOneConsumerFail() {
    int msgSize = 16;

    RMQBroadCastConsumer consumer1 = getBroadCastConsumer(nsAddr, topic, "*",
        new RMQNormalListner());
    RMQBroadCastConsumer consumer2 = getBroadCastConsumer(nsAddr,
        consumer1.getConsumerGroup(), topic, "*",
        new RMQNormalListner(ConsumeConcurrentlyStatus.RECONSUME_LATER));

    producer.send(msgSize);
    Assert.assertEquals("Not all sent succeeded", msgSize, producer.getAllUndupMsgBody().size());

    consumer1.getListner().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);

    assertThat(VerifyUtils.getFilterdMessage(producer.getAllMsgBody(),
        consumer1.getListner().getAllMsgBody()))
        .containsExactlyElementsIn(producer.getAllMsgBody());
}

开发者ID:lirenzuo，项目名称:rocketmq-rocketmq-all-4.1.0-incubating，代码行数:20，代码来源:BroadCastNormalMsgRecvFailIT.java

示例12: main

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException, MQClientException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupNamecc4");

    ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
    File classFile = new File(classLoader.getResource("MessageFilterImpl.java").getFile());

    String filterCode = MixAll.file2String(classFile);
    consumer.subscribe("TopicTest", "org.apache.rocketmq.example.filter.MessageFilterImpl",
        filterCode);

    consumer.registerMessageListener(new MessageListenerConcurrently() {

        @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");
}

开发者ID:lyy4j，项目名称:rmq4note，代码行数:25，代码来源:Consumer.java

示例13: consumeMessage

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
    for (MessageExt msg : msgs) {
        msgIndex.getAndIncrement();
        if (isDebug) {
            if (listnerName != null && listnerName != "") {
                logger.info(listnerName + ":" + msgIndex.get() + ":"
                    + String.format("msgid:%s broker:%s queueId:%s offset:%s",
                    msg.getMsgId(), msg.getStoreHost(), msg.getQueueId(),
                    msg.getQueueOffset()));
            } else {
                logger.info(msg);
            }
        }

        msgBodys.addData(new String(msg.getBody()));
        originMsgs.addData(msg);
        originMsgIndex.put(new String(msg.getBody()), msg);
    }
    return consumeStatus;
}

开发者ID:lyy4j，项目名称:rmq4note，代码行数:22，代码来源:RMQNormalListner.java

示例14: main

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException, MQClientException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("broadcast_consumer_group");
    consumer.setNamesrvAddr("127.0.0.1:9876");

    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

    consumer.setMessageModel(MessageModel.BROADCASTING);

    consumer.subscribe("TopicTest", "TagA || TagC || TagD");

    consumer.registerMessageListener(new MessageListenerConcurrently() {

        @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("Broadcast Consumer Started.%n");
}

开发者ID:javahongxi，项目名称:whatsmars，代码行数:24，代码来源:PushConsumer.java

示例15: main

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException, MQClientException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("push_consumer_group");
    consumer.setNamesrvAddr("127.0.0.1:9876");
    consumer.subscribe("TopicTest", "*");
    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    //wrong time format 2017_0422_221800
    consumer.setConsumeTimestamp("20170422221800");
    consumer.registerMessageListener(new MessageListenerConcurrently() {

        /**

         */
        @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");
}

开发者ID:javahongxi，项目名称:whatsmars，代码行数:22，代码来源:PushConsumer.java

示例16: testConsumer

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
@Test
public void testConsumer() {
    DefaultMQPushConsumer consumer =
            new DefaultMQPushConsumer("PushConsumer");
    consumer.setNamesrvAddr("192.168.146.109:9876");
    try {
        //订阅PushTopic下Tag为push的消息
        consumer.subscribe("PushTopic", "push");
        //程序第一次启动从消息队列头取数据
        consumer.setConsumeFromWhere(
                ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(
                (MessageListenerConcurrently) (list, Context) -> {
                    Message msg = list.get(0);
                    System.out.println(msg.toString());
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
        );
        consumer.start();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

开发者ID:srarcbrsent，项目名称:tc，代码行数:24，代码来源:TcRocketmqTest.java

示例17: main

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException, MQClientException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_1");

    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

    consumer.setMessageModel(MessageModel.BROADCASTING);

    consumer.subscribe("TopicTest", "TagA || TagC || TagD");

    consumer.registerMessageListener(new MessageListenerConcurrently() {

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });

    consumer.start();
    System.out.printf("Broadcast Consumer Started.%n");
}

开发者ID:apache，项目名称:rocketmq，代码行数:23，代码来源:PushConsumer.java

示例18: main

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException, MQClientException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1");
    consumer.subscribe("Jodie_topic_1023", "*");
    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    //wrong time format 2017_0422_221800
    consumer.setConsumeTimestamp("20170422221800");
    consumer.registerMessageListener(new MessageListenerConcurrently() {

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.out.printf("Consumer Started.%n");
}

开发者ID:apache，项目名称:rocketmq，代码行数:18，代码来源:PushConsumer.java

示例19: main

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException, MQClientException, IOException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupNamecc4");

    ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
    File classFile = new File(classLoader.getResource("MessageFilterImpl.java").getFile());

    String filterCode = MixAll.file2String(classFile);
    consumer.subscribe("TopicTest", "org.apache.rocketmq.example.filter.MessageFilterImpl",
        filterCode);

    consumer.registerMessageListener(new MessageListenerConcurrently() {

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });

    consumer.start();

    System.out.printf("Consumer Started.%n");
}

开发者ID:apache，项目名称:rocketmq，代码行数:25，代码来源:Consumer.java

示例20: testPullMessage_Success

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; //导入依赖的package包/类
@Test
public void testPullMessage_Success() throws InterruptedException, RemotingException, MQBrokerException {
    final CountDownLatch countDownLatch = new CountDownLatch(1);
    final MessageExt[] messageExts = new MessageExt[1];
    pushConsumer.getDefaultMQPushConsumerImpl().setConsumeMessageService(new ConsumeMessageConcurrentlyService(pushConsumer.getDefaultMQPushConsumerImpl(), new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            messageExts[0] = msgs.get(0);
            countDownLatch.countDown();
            return null;
        }
    }));

    PullMessageService pullMessageService = mQClientFactory.getPullMessageService();
    pullMessageService.executePullRequestImmediately(createPullRequest());
    countDownLatch.await();
    assertThat(messageExts[0].getTopic()).isEqualTo(topic);
    assertThat(messageExts[0].getBody()).isEqualTo(new byte[] {'a'});
}

开发者ID:apache，项目名称:rocketmq，代码行数:21，代码来源:DefaultMQPushConsumerTest.java

注：本文中的org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。