1.消息消费
public class Consumer {public static void main(String[] args) throws InterruptedException, MQClientException {// 实例化消费者DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");// 设置NameServer的地址consumer.setNamesrvAddr("localhost:9876");// 订阅一个或者多个Topic，以及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);// 标记该消息已经被成功消费return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动消费者实例consumer.start();System.out.printf("Consumer Started.%n");}}
2.顺序消费
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;import org.mon.message.MessageExt;import java.util.List;package org.apache.rocketmq.example.order2;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;import org.mon.consumer.ConsumeFromWhere;import org.mon.message.MessageExt;import java.util.List;import java.util.Random;import java.util.concurrent.TimeUnit;/*** 顺序消息消费，带事务方式（应用可控制Offset什么时候提交）*/public class ConsumerInOrder {public static void main(String[] args) throws Exception {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");consumer.setNamesrvAddr("127.0.0.1:9876");/*** 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>* 如果非第一次启动，那么按照上次消费的位置继续消费*/consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.subscribe("TopicTest", "TagA || TagC || TagD");consumer.registerMessageListener(new MessageListenerOrderly() {Random random = new Random();@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {context.setAutoCommit(true);for (MessageExt msg : msgs) {// 可以看到每个queue有唯一的consume线程来消费, 订单对每个queue(分区)有序System.out.println("consumeThread=" + Thread.currentThread().getName() + "queueId=" + msg.getQueueId() + ", content:" + new String(msg.getBody()));}try {//模拟业务逻辑处理中...TimeUnit.SECONDS.sleep(random.nextInt(10));} catch (Exception e) {e.printStackTrace();}return ConsumeOrderlyStatus.SUCCESS;}});consumer.start();System.out.println("Consumer Started.");}}
3.PushConsumer,PullConsumer消费模式分析
Push:实时性高，但是增加服务器负载，消费能力不同，如果push过快，消费端会出现很多问题；
Pull:消费者从Server端拉取资源，主动权在消费端，可控性好，但是间隔时间不好控制，间隔时间太短，则空请求，资源浪费，间隔时间太长，则消息不能及时处理；
长轮询：Client请求Server服务端（Broker）,Broker会保持一段时间的连接，默认是15s,如果这段时间没有消息达到，则离开返回给Customer,没有消息的话，超过15s,则返回空，再进行重新请求，缺点：服务端需要保持Customer的请求，会占用资源，需要客户端连接数可控，否则会一段连接。
PushConsumer：本质是长轮询；
1.系统收到消息后自动处理消息和offset,如果有新的Consumer加入会自动做负载均衡，
2.在broker端可以通过longPolling=true来开启长轮询，
3.消费端代码：DefaultMQPushConsumerImpl->pullMessage->PullCallback
4.服务端代码：broker.longpolling
5.虽然是push，但代码里面大量使用了pull,是因为使用了长轮询方式达到push效果，既有pull有的，又有push的实现性。
6.关闭优雅：只要是释放资源和保存offset,调用shutdown()即可，参考@PostConstruct,@PreDestory。
PullConsumer:需要自己维护offset;
1.获取MessageQueue遍历；
2.客户维护offset，需要本地用户存储offset,存储内存，磁盘数据库等；
3.处理不同状态的消息FOUND,NO_NEW_MSG,OFFSET_ILLRGL,NO_MATCHED_MSG,4种状态；
4.灵活性强，但编码复杂度高；
5.关闭优雅，注意是释放资源和保存offset,需要程序自己保存offset,特别是异常处理的时候；
