// 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例,ConsumerGroupName需要由应用来保证唯一

final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMQConfig.getConsumerGroup());

consumer.setNamesrvAddr(rocketMQConfig.getNamesrvAddr());

// 订阅指定topic下tags分别等于TagA或TagC或TagD

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

// 订阅指定topic下所有消息，注意：一个consumer对象可以订阅多个topic

consumer.subscribe(rocketMQConfig.getSmsReceiptTopic(), "*");

consumer.setVipChannelEnabled(false);

// 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费 如果非第一次启动，那么按照上次消费的位置继续消费

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

//单线程处理设置最大线程，最小线程也要设置，不然启动报错

consumer.setConsumeThreadMax(1);

consumer.setConsumeThreadMin(1);

//默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息

//MessageListenerOrderly一个线程一个队列顺序接收

consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {

System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

MessageExt msg = (MessageExt) msgs.get(0);

String msgBody = new String(msg.getBody());

//短信回执主题

if (msg.getTopic().equals(rocketMQConfig.getSmsReceiptTopic())) {

log.info("消费者接收到短信回执消息：[{}]", msgBody);

httpSmsSvc.handleSmsReplyAndReport(msgBody);

} else {

log.error("出现意料之外的消息：[{}]", msg.toString());

}

return ConsumeOrderlyStatus.SUCCESS;

});

// Consumer对象在使用之前必须要调用start初始化，初始化一次即可

consumer.start();

System.out.println("Consumer Started.");

Runtime.getRuntime().addShutdownHook(new Thread(consumer::shutdown));

 RocketMQ 版所提供的 TCP Java SDK 支持多线程消费，且适用于所有消息类型，本文介绍如何设置消费线程数的方法。

在启动 Consumer 时，设置一个 ConsumeThreadNums 属性即可。具体示例如下所示。

public static void main(String[] args) {
    Properties properties = new Properties();
    properties.put(PropertyKeyConst.GROUP_ID, "GID_001");
    properties.put(PropertyKeyConst.AccessKey, "xxxxxxxxxxxx");
    properties.put(PropertyKeyConst.SecretKey, "xxxxxxxxxxxx");    /**
     * 设置消费端线程数固定为 20
     */
    properties.setProperty(PropertyKeyConst.ConsumeThreadNums,"20");
    Consumer consumer =ONSFactory.createConsumer(properties);
    consumer.subscribe("TestTopic", "*", new MessageListener() {        public Action consume(Message message, ConsumeContext context) {
            System.out.println("Receive: " + message);            return Action.CommitMessage;
        }
    });
    consumer.start();
    System.out.println("Consumer Started");
}

Rocketmq消费分为push和pull两种方式，push为被动消费类型，pull为主动消费类型，push方式最终还是会从broker中pull消息。不同于pull的是，push首先要注册消费监听器，当监听器处触发后才开始消费消息，所以被称为“被动”消费。

public class PushConsumer {

  

checkConfig 检查内容：

1.消费组 -- （不能与默认DEFAULT_CONSUMER同名）

2.消费模型 -- （默认CLUSTERING）

3.从何处开始消费 -- （默认CONSUME_FROM_LAST_OFFSET）

4.消费时间戳 -- （消息回溯，默认Default backtracking consumption time Half an hour ago）

5.消费负载均衡策略 -- （默认AllocateMessageQueueAveragely）

6.订阅关系 --（map类型，即可订阅多个topic；key=Topic, value=订阅描述）

7.消费监听 --（必须为orderly or concurrently类型之一）

8.消费消息的线程数量控制 -- （消费线程池最大、最小数量）

9.检查单队列并行消费允许的最大跨度 --（consumeConcurrentlyMaxSpan）

10.检查拉消息本地队列缓存消息最大数 --（pullThresholdForQueue）（processQueue.getMsgCount()记数）

11.检查拉取时间间隔 --（拉消息间隔，由于是长轮询，所以默认为0）

12.检查批量消费的个数 --（一次消费多少条消息）

13.检查批量拉取消息的个数 --（一次最多拉多少条）