淘宝消息中间件RocketMQ的安装和简单使用

RocketMQ是什么？
RocketMQ是一款分布式、队列模型的消息中间件，具有以下特点：
能够保证严格的消息顺序
提供丰富的消息拉取模式
高效的订阅者水平扩展能力
实时的消息订阅机制
亿级消息堆积能力
RocketMQ是alibaba开源的java消息中间件。在github上开源，项目同时也在oschina上。地址：https://git.oschina.net/vintagewang/RocketMQ/
下载RocketMQ3.0.8 最新的是3.1.0 试了好多次，编译怎么都不行啊。
果断使用了3.0.8
https://github.com/alibaba/RocketMQ/archive/v3.0.8.zip 本人机器有点问题git居然安装不上（纳闷啊）直接下的zip包。
解压开来，找到项目中pom.xml并且把其中的
<parent>
<groupId>com.taobao</groupId>
<artifactId>parent</artifactId>
<version>1.0.2</version>
</parent>
给注释掉，不然弄死不能编译的，我可是折腾了好久好久啊，这是由于阿里开源不彻底导致的。
下面开始编译吧。cmd控制台进到解压文件夹的pom.xml目录 执行
mvn -Dmaven.test.skip=true clean package install assembly:assembly -U  或者直接运行根目录下的install.cmd 推荐控制台运行，能直观的看到错误信息。install.cmd出错了控制台就没了。注释了parent之后应该就不报错了。等待它去网上下载完的jar包，并且编译完吧。
编译完成之后，根目录下回多出来一个叫target的目录，将里面的alibaba-rocketmq-3.0.8文件夹拷贝到你的安装盘符下吧，
命令行进入到${alibaba-rocketmq-3.0.8}/bin 输入start mqnamesrv.exe 会弹出一个信息窗口，记录的是日志也可以写到文件中 后面接着写 >${你的日志存放目录} ，在日志文件中看到The name Server boot success 说明启动成功了，输入jps -v 检测
2636 -Djava.ext.dirs=F:\taobao\alibaba-rocketmq-3.0.8\alibaba-rocketmq\bin/../lib-Drocketmq.home.dir=F:\taobao\alibaba-rocketmq-3.0.8\alibaba-rocketmq\bin/..-XX:MaxNewSize=512M -XX:MaxPermSize=128M -XX:NewSize=256M -XX:PermSize=128M -Xms512m -Xmx1g exit abort
启动borker start mqbroker.exe -n 127.0.0.1:9876 同样的弹出一个窗口 看到success表示成功了，文件日志同上。
Java操作列子，来自开源项目中。
建立一个普通的maven项目
Pom中加入如下配置

<dependencies>
    <dependency>
         <groupId>com.alibaba.rocketmq</groupId>
         <artifactId>rocketmq-client</artifactId>
         <version>3.0.8</version>
      </dependency>
      <dependency>
         <groupId>com.alibaba.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
         <version>3.0.8</version>
         <type>pom</type>
      </dependency>
      <dependency>
         <groupId>ch.qos.logback</groupId>
         <artifactId>logback-classic</artifactId>
         <version>1.1.1</version>
      </dependency>
      <dependency>
         <groupId>ch.qos.logback</groupId>
         <artifactId>logback-core</artifactId>
         <version>1.1.1</version>
      </dependency>
      <dependency>
         <groupId>junit</groupId>
         <artifactId>junit</artifactId>
         <version>4.10</version>
         <scope>test</scope>
      </dependency>
  </dependencies>

消息生产者

package rocketmq_test.test;

import java.util.concurrent.TimeUnit;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;

public class Producer {
	 public static void main(String[] args) throws MQClientException,
     InterruptedException{
  /**
   * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br>
   * 注意：ProducerGroupName需要由应用来保证唯一<br>
   * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
   * 因为服务器会回查这个Group下的任意一个Producer
   */
  final DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
  producer.setNamesrvAddr("127.0.0.1:9876");
  producer.setInstanceName("Producer");

  /**
   * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br>
   * 注意：切记不可以在每次发送消息时，都调用start方法
   */
  producer.start();

  /**
   * 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。
   * 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态，<br>
   * 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，<br>
   * 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。
   */
  for (int i = 0; i < 10; i++){
     try {
        {
            Message msg = new Message("TopicTest1",// topic
                  "TagA",// tag
                  "OrderID001",// key
                  ("Hello MetaQA").getBytes());// body
            SendResult sendResult = producer.send(msg);
            System.out.println(sendResult);
        }

        {
            Message msg = new Message("TopicTest2",// topic
                  "TagB",// tag
                  "OrderID0034",// key
                  ("Hello MetaQB").getBytes());// body
            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.MILLISECONDS.sleep(1000);
  }

  /**
   * 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己
   * 注意：我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法
   */
//producer.shutdown();
  Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
     public void run() {
        producer.shutdown();
     }
  }));
  System.exit(0);
}
}

消费者处理消息

package rocketmq_test.test;

import java.util.List;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;

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.");  
    }  
}

启动消费者
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-8 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-9 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-10 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-11 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-12 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-13 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-14 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-15 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-16 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-17 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-18 Receive New Messages: 1
Hello MetaQB
ConsumeMessageThread-ConsumerGroupName-19 Receive New Messages: 1
Hello MetaQA
ConsumeMessageThread-ConsumerGroupName-20 Receive New Messages: 1
Hello MetaQB

运行生产者，
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002715, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=2], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F00000000000027AB, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=2], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002840, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F00000000000028D5, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F000000000000296B, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=4]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002A00, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002A95, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002B2B, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002BC0, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002C55, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002CEB, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=7]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002D80, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=2], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002E15, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=2], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002EAB, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=2], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002F40, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000002FD5, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F000000000000306B, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=3], queueOffset=5]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000003100, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000003195, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F000000000000322B, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=0], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F00000000000032C0, messageQueue=MessageQueue [topic=TopicTest1, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F0000000000003355, messageQueue=MessageQueue [topic=TopicTest2, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=8]
SendResult [sendStatus=SEND_OK, msgId=C0A8490100002A9F00000000000033EB, messageQueue=MessageQueue [topic=TopicTest3, brokerName=E97M3HS2ANMGCSZ, queueId=1], queueOffset=8]

评论

3 楼 a6186694 2016-11-05  

247687009 写道

aiyoaiyo0330 写道

请教个问题 我按照你的代码 写了一遍 但是报错 com.alibaba.rocketmq.client.exception.MQClientException: No route info of this topic, TopicTest1   我把mqnamesrv.exe 开启了 我是在windows下运行的

你可以去看看淘宝开源

不能直接启动mqbroker.exe ；
得这样子  开启另一个windows终端cmd，进入解压的bin目录，也可一步输入mqbroker -n 127.0.0.1:9876启动broker，保持mqbroker.exe运行，不要关闭这个终端;然后把这个9876端口号写在 consumer.setNamesrvAddr("127.0.0.1:9876"); 

2 楼 yjq8116 2015-01-02  

247687009 写道

aiyoaiyo0330 写道

请教个问题 我按照你的代码 写了一遍 但是报错 com.alibaba.rocketmq.client.exception.MQClientException: No route info of this topic, TopicTest1   我把mqnamesrv.exe 开启了 我是在windows下运行的

你可以去看看淘宝开源

这个问题我也遇到了，这是因为启动broker需要 mqbroker.exe -n localhost:9876，而不是双击 mqbroker.exe

1 楼 247687009 2014-06-13  

aiyoaiyo0330 写道

请教个问题 我按照你的代码 写了一遍 但是报错 com.alibaba.rocketmq.client.exception.MQClientException: No route info of this topic, TopicTest1   我把mqnamesrv.exe 开启了 我是在windows下运行的

你可以去看看淘宝开源