Rocketmq消费者示例

示例"/>

Rocketmq消费者示例

        上一篇讲述了生产者示例，RocketMQ入门示例-生产者-CSDN博客

        这里继续消费者示例

拉模式

public class ConsumerPull {public static void main(String[] args) throws MQClientException {DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("group_one");consumer.setNamesrvAddr("127.0.0.1:9800");consumer.start();try {MessageQueue mq = new MessageQueue();mq.setQueueId(0);mq.setTopic("TopicTest");mq.setBrokerName("broker-a");long offset = 0;PullResult pullResult = consumer.pull(mq, "*", offset, 32);if (pullResult.getPullStatus().equals(PullStatus.FOUND)) {System.out.println(pullResult.getMsgFoundList().size());System.out.printf("%s%n", pullResult.getMsgFoundList());consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());}} catch (Exception e) {e.printStackTrace();}consumer.shutdown();}
}

推模式

public class ConsumerPush {public static void main(String[] args) throws InterruptedException, MQClientException {// 初始化consumer，并设置consumer group nameDefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_push");// 设置NameServer地址consumer.setNamesrvAddr("127.0.0.1:9800");//订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息consumer.subscribe("TopicTest", "*");//注册回调接口来处理从Broker中收到的消息consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);// 返回消息消费状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动Consumerconsumer.start();System.out.printf("Consumer Started.%n");}}

轻量模式

       Lite Pull Consumer是RocketMQ 4.6.0推出的Pull Consumer，相比于原始的Pull Consumer更加简单易用， 它提供了Subscribe和Assign两种模式

public class LitePullConsumerSubscribe {public static volatile boolean running = true;public static void main(String[] args) throws Exception {DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("lite_pull_consumer_test");litePullConsumer.subscribe("TopicTest", "*");litePullConsumer.setPullBatchSize(20);litePullConsumer.start();try {while (running) {List<MessageExt> messageExts = litePullConsumer.poll();System.out.printf("%s%n", messageExts);}} finally {litePullConsumer.shutdown();}}// 如下是Assign模式的示例
//    public static void main(String[] args) throws Exception {
//        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("please_rename_unique_group_name");
//        litePullConsumer.setAutoCommit(false);
//        litePullConsumer.start();
//        Collection<MessageQueue> mqSet = litePullConsumer.fetchMessageQueues("TopicTest");
//        List<MessageQueue> list = new ArrayList<>(mqSet);
//        List<MessageQueue> assignList = new ArrayList<>();
//        for (int i = 0; i < list.size() / 2; i++) {
//            assignList.add(list.get(i));
//        }
//        litePullConsumer.assign(assignList);
//        litePullConsumer.seek(assignList.get(0), 10);
//        try {
//            while (running) {
//                List<MessageExt> messageExts = litePullConsumer.poll();
//                System.out.printf("%s %n", messageExts);
//                litePullConsumermitSync();
//            }
//        } finally {
//            litePullConsumer.shutdown();
//        }
//    }
}

完整代码：Xu Qian/rocketmq-client