package?com.kang.rocketmq;
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.apache.rocketmq.client.exception.MQClientException;
import?org.apache.rocketmq.common.message.MessageExt;
?
import?java.util.List;
?
public?class?Consumer?{
?
????/**
?????*?當前例子是PushConsumer用法，使用方式給用戶感覺是消息從RocketMQ服務(wù)器推到了應(yīng)用客戶端。

?????*?但是實際PushConsumer內(nèi)部是使用長輪詢Pull方式從metaQ服務(wù)器拉消息，然后再回調(diào)用戶Listener方法

?????*/
????public?static?void?main（String[]?args）?throws?InterruptedException
????????????MQClientException?{
????????/**
?????????*?一個應(yīng)用創(chuàng)建一個Consumer，由應(yīng)用來維護此對象，可以設(shè)置為全局對象或者單例

?????????*?注意：ConsumerGroupName需要由應(yīng)用來保證唯一
?????????*/
????????DefaultMQPushConsumer?consumer?=?new?DefaultMQPushConsumer（
????????????????“ConsumerGroupName“）;
????????consumer.setNamesrvAddr（“127.0.0.1:9876“）;
????????consumer.setInstanceName（“Consumber“）;
?
????????/**
?????????*?訂閱指定topic下tags分別等于TagA或TagC或TagD
?????????*/
????????consumer.subscribe（“TopicTest1“?“TagA?||?TagC?||?TagD“）;
????????/**
?????????*?訂閱指定topic下所有消息

?????????*?注意：一個consumer對象可以訂閱多個topic
?????????*/
????????consumer.subscribe（“TopicTest2“?“*“）;
?
????????consumer.registerMessageListener（new?MessageListenerConcurrently（）?{
?
????????????/**
?????????????*?默認msgs里只有一條消息，可以通過設(shè)置consumeMessageBatchMaxSize參數(shù)來批量接收消息
?????????????*/
????????????@Override
????????????public?ConsumeConcurrentlyStatus?consumeMessage（
????????????????????List?msgs?ConsumeConcurrentlyContext?context）?{
????????????????System.out.println（Thread.currentThread（）.getName（）
????????????????????????+?“?Receive?New?Messages:?“?+?msgs.size（））;
????????????????MessageExt?msg?=?msgs.get（0）;
????????????????if?（msg.getTopic（）.equals（“TopicTest1“））?{
????????????????????//?執(zhí)行TopicTest1的消費邏輯
????????????????????if?（msg.getTags（）?!=?null?&&?msg.getTags（）.equals（“TagA“））?{
????????????????????????//?執(zhí)行TagA的消費
????????????????????????System.out.println（new?String（msg.getBody（）））;
????????????????????}?else?if?（msg.getTags（）?!=?null
????????????????????????????&&?msg.getTags（）.equals（“TagC“））?{
????????????????????????//?執(zhí)行TagC的消費
????????????????????}?else?if?（msg.getTags（）?!=?null
????????????????????????????&&?msg.getTags（）.equals（“TagD“））?{
????????????????????????//?執(zhí)行TagD的消費
????????????????????}
????????????????}?else?if?（msg.getTopic（）.equals（“TopicTest2“））?{
????????????????????System.out.println（new?String（msg.getBody（）））;
????????????????}
?
????????????????return?ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
????????????}
????????}）;
?
????????/**
?????????*?Consumer對象在使用之前必須要調(diào)用start初始化，初始化一次即可

?????????*/
????????consumer.start（）;
?
????????System.out.println（“Consumer?Started.“）;
????}
}

?屬性????????????大小?????日期????時間???名稱
-----------?---------??----------?-----??----

?????文件???????1433??2019-03-26?16:29??RocketMQ\.classpath

?????文件????????560??2019-03-26?15:04??RocketMQ\.project

?????文件????????191??2019-03-26?16:29??RocketMQ\.settings\org.eclipse.core.resources.prefs

?????文件????????243??2019-03-26?16:29??RocketMQ\.settings\org.eclipse.jdt.core.prefs

?????文件?????????90??2019-03-26?15:04??RocketMQ\.settings\org.eclipse.m2e.core.prefs

?????文件???????1485??2019-03-26?16:30??RocketMQ\pom.xml

?????文件???????3414??2019-03-26?16:33??RocketMQ\src\main\java\com\kang\rocketmq\Consumer.java

?????文件???????3562??2019-03-26?16:32??RocketMQ\src\main\java\com\kang\rocketmq\Producer.java

?????文件???????2571??2019-03-26?16:33??RocketMQ\target\classes\com\kang\rocketmq\Consumer$1.class

?????文件???????1381??2019-03-26?16:33??RocketMQ\target\classes\com\kang\rocketmq\Consumer.class

?????文件???????2202??2019-03-26?16:32??RocketMQ\target\classes\com\kang\rocketmq\Producer.class

?????目錄??????????0??2019-03-26?15:15??RocketMQ\src\main\java\com\kang\rocketmq

?????目錄??????????0??2019-03-26?15:10??RocketMQ\src\main\java\com\kang

?????目錄??????????0??2019-03-26?16:30??RocketMQ\target\classes\com\kang\rocketmq

?????目錄??????????0??2019-03-26?15:10??RocketMQ\src\main\java\com

?????目錄??????????0??2019-03-26?16:30??RocketMQ\target\classes\com\kang

?????目錄??????????0??2019-03-26?15:10??RocketMQ\src\main\java

?????目錄??????????0??2019-03-26?15:04??RocketMQ\src\main\resources

?????目錄??????????0??2019-03-26?15:04??RocketMQ\src\test\java

?????目錄??????????0??2019-03-26?15:04??RocketMQ\src\test\resources

?????目錄??????????0??2019-03-26?16:30??RocketMQ\target\classes\com

?????目錄??????????0??2019-03-26?15:04??RocketMQ\src\main

?????目錄??????????0??2019-03-26?15:04??RocketMQ\src\test

?????目錄??????????0??2019-03-26?16:30??RocketMQ\target\classes

?????目錄??????????0??2019-03-26?15:04??RocketMQ\target\test-classes

?????目錄??????????0??2019-03-26?16:29??RocketMQ\.settings

?????目錄??????????0??2019-03-26?15:04??RocketMQ\src

?????目錄??????????0??2019-03-26?15:04??RocketMQ\target

?????目錄??????????0??2019-03-26?15:04??RocketMQ

-----------?---------??----------?-----??----

............此處省略2個文件信息