RocketMq学习笔记【原创】

前段时间在实施分布式数据库的过程中，使用过rocketMQ, 因为搭建过程实在太顺利，当时忘了留文档，今天有同事来向我请教rocketMQ使用时，居然已经找不出当时的demo了 。
于是在收藏夹里重新翻出原来的学习内容，干脆写一篇学习笔记，加深映象吧。
1.下载最新版 rocketMq.
[urlhttps://github.com/alibaba/RocketMQ/releases/download/v3.2.6/alibaba-rocketmq-3.2.6.tar.gz[/url]
2.解压，并进入bin目录，运行启动命令：
启动nameserver: Start /b mqnamesrv.exe > mqnamesrv.log
启动broker: Start /b mqbroker.exe –n “192.168.180.15:9876” > mqbroker.log
3.搭建测试工程，引进jar包
<dependency>
      <groupId>com.alibaba.rocketmq</groupId>
      <artifactId>rocketmq-client</artifactId>
      <version>3.2.6</version>
</dependency>
   <dependency>
      <groupId>com.alibaba.rocketmq</groupId>
      <artifactId>rocketmq-tools</artifactId>
      <version>3.2.6</version>
    </dependency>
    <dependency>
      <groupId>com.alibaba.rocketmq</groupId>
      <artifactId>rocketmq-all</artifactId>
      <version>3.2.6</version>
      <type>pom</type>
    </dependency>
4. 编写测试代码
发送端

package cn.gov.zjport.rocketmq.example;

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 {
	 
	/**
	 * 主函数
	 * @param args  参数
	 * @throws MQClientException
	 * @throws InterruptedException
	 */
	public static void main(String[] args) throws MQClientException,  
     InterruptedException{  
		  /**  
		   * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例<br>  
		   * 注意：ProducerGroupName需要由应用来保证唯一<br>  
		   * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，  
		   * 因为服务器会回查这个Group下的任意一个Producer  
		   */  
		  final DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");  
		  producer.setNamesrvAddr("192.168.180.15: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方法  
			   */  
			  Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {  
			     public void run() {  
			        producer.shutdown();  
			     }  
			  }));  
			  System.exit(0);  
			}  
}  

消费端
package cn.gov.zjport.rocketmq.example;

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 PushConsumer {
/**
     * 当前例子是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("192.168.180.15: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."); 
    } 
}