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}
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.

- 顺序消费:MessageListenerOrderly
- 并发消费: MessageListenerConcurrently
1.
2.

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.
2.
3.
4.
5.
6.
7.
8.

1 package org.apache.rocketmq.client.consumer.listener;
 2
 3/**
 4 * 并发消费mq消息结果
 5 */
 6public enum ConsumeConcurrentlyStatus {
 7
 8    /**
 9     * Success consumption
10     * 成功消费
11     */
12    CONSUME_SUCCESS,
13
14    /**
15     * Failure consumption,later try to consume
16     * 失败消费，稍后尝试消费
17     *
18     *
19     * 如果 {@link MessageListener}返回的消费结果为 RECONSUME_LATER,则需要将这些消息发送给Broker延迟消息。
20     * 如果给broker发送消息失败,将延迟5s后提交线程池进行消费。
21     *
22     * RECONSUME_LATER的消息发送入口: MQClientAPIImpl#consumerSendMessageBack,
23     * 命令编码: {@link org.apache.rocketmq.common.protocol.RequestCode#CONSUMER_SEND_MSG_BACK}
24     */
25    RECONSUME_LATER;
26}
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.

1public synchronized void start() throws MQClientException {
 2        switch (this.serviceState) {
 3            case CREATE_JUST:
 4
 5                // 省略一部分代码...........
 6
 7                // 根据注册的监听器类型[并发消息监听器/顺序执行消息监听器],来确定使用哪种消费服务.   
 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.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.

1// 无界队列,并且不可配置容量.那 DefaultMQPushConsumer#consumeThreadMax 配置就毫无意义了.
2this.consumeRequestQueue = new LinkedBlockingQueue<Runnable>();
3this.consumeExecutor = new ThreadPoolExecutor(
4    this.defaultMQPushConsumer.getConsumeThreadMin(), // 默认20
5    this.defaultMQPushConsumer.getConsumeThreadMax(), // 默认64
6    1000 * 60,
7    TimeUnit.MILLISECONDS,
8    this.consumeRequestQueue,
9    new ThreadFactoryImpl("ConsumeMessageThread_"));
1.
2.
3.
4.
5.
6.
7.
8.
9.

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    }
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.