1public class Consumer {
 2
 3    public static void main(String[] args) throws InterruptedException, MQClientException {
 4        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_consumer_group");
 5        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 6        consumer.subscribe("TopicTest", "*");
 7        consumer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
 8        consumer.registerMessageListener(new MessageListenerConcurrently() {
 9            @Override
10            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
11                ConsumeConcurrentlyContext context) {
12                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
13                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
14            }
15        });
16        consumer.start();
17        System.out.printf("Consumer Started.%n");
18    }
19}

- 順序消費(fèi):MessageListenerOrderly
- 并發(fā)消費(fèi): MessageListenerConcurrently

1 consumer.registerMessageListener(new MessageListenerConcurrently() {
2        @Override
3        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
4            ConsumeConcurrentlyContext context) {
5            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
6            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
7        }
8    });

1 package org.apache.rocketmq.client.consumer.listener;
 2
 3/**
 4 * 并發(fā)消費(fèi)mq消息結(jié)果
 5 */
 6public enum ConsumeConcurrentlyStatus {
 7
 8    /**
 9     * Success consumption
10     * 成功消費(fèi)
11     */
12    CONSUME_SUCCESS,
13
14    /**
15     * Failure consumption,later try to consume
16     * 失敗消費(fèi)，稍后嘗試消費(fèi)
17     *
18     *
19     * 如果 {@link MessageListener}返回的消費(fèi)結(jié)果為 RECONSUME_LATER,則需要將這些消息發(fā)送給Broker延遲消息。
20     * 如果給broker發(fā)送消息失敗,將延遲5s后提交線程池進(jìn)行消費(fèi)。
21     *
22     * RECONSUME_LATER的消息發(fā)送入口: MQClientAPIImpl#consumerSendMessageBack,
23     * 命令編碼: {@link org.apache.rocketmq.common.protocol.RequestCode#CONSUMER_SEND_MSG_BACK}
24     */
25    RECONSUME_LATER;
26}

1public synchronized void start() throws MQClientException {
 2        switch (this.serviceState) {
 3            case CREATE_JUST:
 4
 5                // 省略一部分代碼...........
 6
 7                // 根據(jù)注冊(cè)的監(jiān)聽(tīng)器類型[并發(fā)消息監(jiān)聽(tīng)器/順序執(zhí)行消息監(jiān)聽(tīng)器],來(lái)確定使用哪種消費(fèi)服務(wù).   
 8                if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
 9                    this.consumeOrderly = true;
10                    this.consumeMessageService = new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
11                } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
12                    this.consumeOrderly = false;
13                    this.consumeMessageService = new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
14                }
15                this.consumeMessageService.start();
16
17                // 省略一部分代碼..........
18                this.serviceState = ServiceState.RUNNING;
19                break;
20            case RUNNING:
21            case START_FAILED:
22            case SHUTDOWN_ALREADY:
23                throw new MQClientException("The PushConsumer service state not OK, maybe started once");
24            default:
25                break;
26        }
27
28        // 省略一部分代碼..........
29    }

1// 無(wú)界隊(duì)列,并且不可配置容量.那 DefaultMQPushConsumer#consumeThreadMax 配置就毫無(wú)意義了.
2this.consumeRequestQueue = new LinkedBlockingQueue<Runnable>();
3this.consumeExecutor = new ThreadPoolExecutor(
4    this.defaultMQPushConsumer.getConsumeThreadMin(), // 默認(rèn)20
5    this.defaultMQPushConsumer.getConsumeThreadMax(), // 默認(rèn)64
6    1000 * 60,
7    TimeUnit.MILLISECONDS,
8    this.consumeRequestQueue,
9    new ThreadFactoryImpl("ConsumeMessageThread_"));

1public static void main(String[] args) throws InterruptedException, MQClientException {
 2        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
 3
 4        consumer.setConsumeThreadMin(30);
 5        consumer.setConsumeThreadMax(64);
 6
 7        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 8        consumer.subscribe("TopicTest", "*");
 9        consumer.registerMessageListener(new MessageListenerConcurrently() {
10
11            @Override
12            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
13                ConsumeConcurrentlyContext context) {
14                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
15                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
16            }
17        });
18        consumer.start();
19        System.out.printf("Consumer Started.%n");
20    }