《RocketMQ 安裝和使用》

安裝Maven

安裝步驟：《Maven的安裝、配置及使用入門》

　　http://www.cnblogs.com/dcba1112/archive/2011/05/01/2033805.html

　　http://maven.apache.org/download.cgi （apache-maven-3.3.3-bin.zip）

　　Path環境變量，當我們在cmd中輸入命令時，Windows首先會在當前目錄中尋找可執行文件或腳本，如果沒有找到，Windows會接着遍歷環境變量Path中定義的路徑。由於我們將%M2_HOME%\bin添加到了Path中，而這里%M2_HOME%實際上是引用了我們前面定義的另一個變量，其值是Maven的安裝目錄。因此，Windows會在執行命令時搜索目錄D:\bin\apache-maven-3.0\bin，而mvn執行腳本的位置就是這里。

 

安裝RocketMQ

源碼地址：(2015-07-18 3.2.6版本)

　　E:\RocketMQ\RocketMQ-master

　　E:\RocketMQ\RocketMQ_Workspace

　　E:\RocketMQ\RocketMQ-master\target\alibaba-rocketmq-3.2.6-alibaba-rocketmq\alibaba-rocketmq\bin

 

執行下邊的命令或者執行install.bat（在這個bat文件中的命令如下）對maven熟悉的一眼就知道是執行clean package install assembly等操作。

　　mvn -Dmaven.test.skip=true clean packageinstall assembly:assembly –U

 

將RocketMQ-master導入到eclipse中

　　　　

　　進入剛生成的target文件夾下的bin目錄，在命令行中執行：start mqnamesrv.exe，會彈出一個信息窗口，顯示The name Server boot success 說明啟動成功了，接着啟動borker，在命令行中執行：start mqbroker.exe -n 127.0.0.1:9876 同樣的彈出一個窗口，看到success表示成功了。

　　start mqnamesrv >E:\RocketMQ\logs\alibaba-rocketmq/mqnamesrv.log

【遇到問題】無法啟動mqnamesrv。顯示軟件不兼容。

　　

 

【如何在windows下使用linux的shell腳本】
　　http://www.cygwin.com/。

　　E:\cygwin64。

【當前目錄調出CMD】

　　在桌面上先按住Shift鍵，然后鼠標右鍵，出現選項“在此處打開命令窗口（W）”也可以打開命令行。

生產者

public class Producer{
    public static void main(String[] args) throws MQClientException,InterruptedException{
        //一個應用創建一個Producer，由應用來維護此對象，可以設置為全局對象或者單例
        final DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        producer.setNamesevAddr("127.0.0.1:9876");
        producer.setInstanceName("Producer");
        
        producer.start();
        
        //一個Producer對象可以發送多個topic，多個tag的消息
        for(int i=0; i<10; i++){
            try{
                {
                    Message msg = new Message("TopicTest1","TagA","OrderID001",("Hello MetaQ").getBytes());
                    SendResult sendResult = pro .send(msg);
                    System.out.println(sendResult);
                }
                {
                    Message msg = new Message("TopicTest2", "TagB","OrderID0034",("Hello MetaQB").getBytes());
                    SendResult sendResult = producer.send(msg);    
                    System.out.println(sendResult);  
                }
                {    
                    Message msg = new Message("TopicTest3",// topic    
                        "TagC",// tag    
                        "OrderID061",// key    
                        ("Hello MetaQC").getBytes());// body    
                    SendResult sendResult = producer.send(msg);    
                    System.out.println(sendResult);    
                }   
            }catch(Exception e){
                e.printStackTrace();
            }
            TimeUnit.MILLSECONDS.sleep(1000);
        }
        
      /**  
   * 應用退出時，要調用shutdown來清理資源，關閉網絡連接，從MetaQ服務器上注銷自己  
   * 注意：我們建議應用在JBOSS、Tomcat等容器的退出鈎子里調用shutdown方法  
   */    
//producer.shutdown();    
  Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {    
     public void run() {    
        producer.shutdown();    
     }    
  }));    
  System.exit(0);      
    }

}

消費者

public class Consumer {    
     /**   
     * 當前例子是PushConsumer用法，使用方式給用戶感覺是消息從RocketMQ服務器推到了應用客戶端。<br>   
     * 但是實際PushConsumer內部是使用長輪詢Pull方式從MetaQ服務器拉消息，然后再回調用戶Listener方法<br>   
     */      
    public static void main(String[] args) throws InterruptedException,      
                       MQClientException{      
              /**   
               * 一個應用創建一個Consumer，由應用來維護此對象，可以設置為全局對象或者單例<br>   
               * 注意：ConsumerGroupName需要由應用來保證唯一   
               */      
              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下所有消息<br>   
               * 注意：一個consumer對象可以訂閱多個topic   
               */      
              consumer.subscribe("TopicTest2","*");      
    
              consumer.registerMessageListener(new MessageListenerConcurrently() {      
    
                       public ConsumeConcurrentlyStatus consumeMessage(      
                                          List<MessageExt>msgs, ConsumeConcurrentlyContext context) {      
    
                                System.out.println(Thread.currentThread().getName()      
                                                   +" Receive New Messages: " + msgs.size());      
    
                                MessageExt msg = msgs.get(0);      
                                if(msg.getTopic().equals("TopicTest1")) {      
                                          //執行TopicTest1的消費邏輯      
                                          if(msg.getTags() != null && msg.getTags().equals("TagA")) {      
                                                   //執行TagA的消費      
                                                   System.out.println(new String(msg.getBody()));      
                                          }else if (msg.getTags() != null      
                                                            &&msg.getTags().equals("TagC")) {      
                                                   //執行TagC的消費      
                                                   System.out.println(new String(msg.getBody()));      
                                          }else if (msg.getTags() != null      
                                                            &&msg.getTags().equals("TagD")) {      
                                                   //執行TagD的消費      
                                                   System.out.println(new String(msg.getBody()));      
                                          }      
                                }else if (msg.getTopic().equals("TopicTest2")) {      
                                          System.out.println(new String(msg.getBody()));      
                                }      
    
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;      
    
                       }      
              });      
    
              /**   
               * Consumer對象在使用之前必須要調用start初始化，初始化一次即可<br>   
               */      
              consumer.start();      
    
              System.out.println("ConsumerStarted.");      
    }      
}    

 

參考文章

l  《RocketMQ在windows上安裝和開發使用》http://blog.csdn.net/ruishenh/article/details/22390809

l  《RocketMQ在Windows平台下環境搭建》http://www.cnblogs.com/marcotan/p/4256859.html

l  《RocketMQ隨筆分類》http://www.cnblogs.com/marcotan/category/655319.html