`

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."); 
    } 
}
分享到:
评论
发表评论

文章已被作者锁定,不允许评论。

相关推荐

    RocketMQ学习笔记 1

    RocketMQ学习笔记 1 RocketMQ学习笔记 1是关于Apache RocketMQ的学习笔记,涵盖了RocketMQ的基本概念、架构、消息模型、Producer和Consumer的基本概念、JMS规范等知识点。 1. 消息模型 在RocketMQ中,消息模型是...

    RocketMQ学习笔记 .pdf

    RocketMQ学习笔记主要涵盖了消息队列(MQ)的相关知识,内容深入浅出,不仅包含理论知识,还融入了实战案例和源码分析,适用于希望掌握RocketMQ的开发者。 首先,RocketMQ是一个分布式、开源的消息中间件,它起源于...

    RocketMQ学习笔记1

    RocketMQ学习笔记1 RocketMQ是Apache旗下的一个开源的消息队列系统,具有分布式、可靠、可扩展、高性能等特点。下面是对RocketMQ的学习笔记的总结。 分布式架构 RocketMQ原生支持分布式,解决了单点故障问题,...

    rocketmq学习笔记ppt

    rocketmq学习 rocketmq深入

    尚硅谷完整的关于rocketmq的学习视频整理笔记

    尚硅谷提供的 RocketMQ 学习视频笔记旨在帮助初学者系统地掌握 RocketMQ 的核心概念和使用方法。 1. **MQ 简述** 消息队列(Message Queue,MQ)是一种异步通信机制,它在生产者和消费者之间充当了缓冲的角色,...

    RocketMq 学习笔记

    RocketMq 学习笔记 包含集群配置、顺序消费、事务、Filter组件配置以及使用,包含demo

    RocketMQ黑马笔记

    "RocketMQ黑马笔记" RocketMQ是阿里巴巴开源的消息中间件,在阿里内部历经了双十一等很多高并发场景的考验,能够处理亿万级别的消息。2016年开源后捐赠给Apache,现在是Apache的一个顶级项目。目前RocketMQ在阿里云...

    rocketmq学习笔记---Docker Desktop下安装rocketmq.md

    Docker Desktop下安装rocketmq

    rocketmq自用笔记1

    rocketmq自用笔记第一部分

    Java学习笔记学习笔记

    Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习笔记Java学习...

    希沃白板学习笔记.pdf

    希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf希沃白板学习笔记.pdf

    Springcloud学习笔记.md

    Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Springcloud学习笔记.md,Spring...

    人工智能学习笔记,人工智能学习笔记

    人工智能学习笔记,人工智能学习笔记,人工智能学习笔记人工智能学习笔记,人工智能学习笔记,人工智能学习笔记人工智能学习笔记,人工智能学习笔记,人工智能学习笔记人工智能学习笔记,人工智能学习笔记,人工智能...

    CCNA学习笔记 CCNA学习笔记

    CCNA学习笔记 CCNA学习笔记 CCNA学习笔记

    云的学习笔记-云的学习笔记系统-云的学习笔记系统源码-云的学习笔记管理系统-基于ssm的云的学习笔记系统-ssm-java代码

    云的学习笔记-云的学习笔记系统-云的学习笔记系统源码-云的学习笔记管理系统-云的学习笔记管理系统java代码-云的学习笔记系统设计与实现-基于ssm的云的学习笔记系统-基于Web的云的学习笔记系统设计与实现-云的学习...

    云的学习笔记-云的学习笔记系统-云的学习笔记系统源码-云的学习笔记管理系统-基于Web的云的学习笔记系统设计与实现-java代码

    云的学习笔记-云的学习笔记系统-云的学习笔记系统源码-云的学习笔记管理系统-云的学习笔记管理系统java代码-云的学习笔记系统设计与实现-基于ssm的云的学习笔记系统-基于Web的云的学习笔记系统设计与实现-云的学习...

Global site tag (gtag.js) - Google Analytics