SpringBoot整合阿里云rocketmq消息队列，创建生产者和消费者实例

素颜马尾好姑娘i 2022-03-22 02:37 2162阅读 0赞

### 简介 ###

*  **消息队列（Message Queue，简称 MQ）**是构建分布式互联网应用的基础设施，通过 MQ 实现的松耦合架构设计可以提高系统可用性以及可扩展性，是适用于现代应用的最佳设计方案；
 *  **RocketMQ**，是阿里巴巴自研消息中间件产品，使用它有**两种方式**：

1.   自行搭建rocketmq集群，然后创建rocketmq启动包、生产者和消费者，进行消息推送和消费。这种方式在自己拥有可靠服务器的前提下，可以自己搭建属于自己的rocketmq集群，但是如果服务器性能不太好的话，集群很容易宕机，造成消息丢失的情况；
2.  直接购买阿里云的rocketmq产品，然后利用阿里云提供的demo，创建生产者和消费者，进行消息推送和消费。这种方式价格便宜，进入阿里云控制台维护也相对方便，普遍应用于中小型公司；

*  接入rocketm生产消息有两种普遍方式，**一种是tcp，另一种是mqtt**，大部分情况下都是使用tcp接入，而如果是安卓、ios等终端接入的话，一般使用mqtt接入，然后用rocketmq服务器消费，详细的对比可链接官网：[https://help.aliyun.com/document\_detail/94521.html?spm=a2c4g.11186623.6.543.18a07871mTGzwC][https_help.aliyun.com_document_detail_94521.html_spm_a2c4g.11186623.6.543.18a07871mTGzwC]
 *  本篇博客主要讲述，使用springboot框架接入阿里云rocketmq产品，然后创建生产者和消费者，进行消息推送和消费。由于官网给的demo集合了太多没有用到的实例和方法，所以就抽取部分出来，进行整合；

### 实践过程 ###

*  购买阿里云产品，然后依据官网流程，创建topic，然后根据主题创建生产者id和消费者id（**2019年1月23号阿里云控制台的rocketmq版本更新之后，生产者id和消费者id统一合并为groupid，不用创建复杂的连接关系，接入更加简单**），同时获取账号的accessKey和secretKey；
 *  **创建生产者，生产消息**（生产者和消费者可合并成一个项目，这里不做合并）：

1. 引入依赖包：

<dependency>
    	<groupId>com.aliyun.openservices</groupId>
    	<artifactId>ons-client</artifactId>
    	<version>1.7.8.Final</version>
    </dependency>

2. yml配置文件添加属性：

#配置rocketmq      
    rocketmq:
      producer:
    	producerId: GroupId_Test #生产者id(旧版本是生产者id，新版本是groupid)，替换成自己的
    	msgTopic: Test #生产主题，替换成自己的
    	accessKey: XXX  #连接通道，替换成自己的
    	secretKey: XXX  #连接秘钥，替换成自己的
    	onsAddr: http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet  #生产者ons接入域名，替换成自己的

**3. 初始化生产者：**

package com.sixmonth.rocketmq.common.rocketmq.init;
    
    import java.util.Properties;
    
    import javax.annotation.PostConstruct;
    
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    import com.aliyun.openservices.ons.api.ONSFactory;
    import com.aliyun.openservices.ons.api.Producer;
    import com.aliyun.openservices.ons.api.PropertyKeyConst;
    
    /**
     * rocketmq生产者启动初始化类
     * @author sixmonth
     * @Date 2019年1月17日
     *
     */
    @Component
    public class RocketmqProducerInit {
    	
    	  @Value("${rocketmq.producer.producerId}")
    	  private String producerId;
    	 
    	  @Value("${rocketmq.producer.accessKey}")
    	  private String accessKey;
    	 
    	  @Value("${rocketmq.producer.secretKey}")
    	  private String secretKey;
    	 
    	  @Value("${rocketmq.producer.onsAddr}")
    	  private String ONSAddr;
    	 
    	  private static Producer producer;
    	 
    	  /* 
    	  
    	  //当无法注入实例的时候可以使用此方法进行实例初始化
    	  private static class ProducerHolder {
    	    private static final RocketmqProducerInit INSTANCE = new RocketmqProducerInit();
    	  }
    	 
    	  private RocketmqProducerInit (){
    		  
    	  }
    	 
    	  public static final RocketmqProducerInit getInstance() {
    	    return ProducerHolder.INSTANCE;
    	  }*/
    	 
    	  @PostConstruct
    	  public void init(){
    		System.out.println("初始化启动生产者！");
    	    // producer 实例配置初始化
    	    Properties properties = new Properties();
    	    //您在控制台创建的Producer ID
    	    properties.setProperty(PropertyKeyConst.GROUP_ID, producerId);
    	    // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
    	    properties.setProperty(PropertyKeyConst.AccessKey, accessKey);
    	    // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
    	    properties.setProperty(PropertyKeyConst.SecretKey, secretKey);
    	    //设置发送超时时间，单位毫秒
    	    properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
    	    // 设置 TCP 接入域名(此处以公共云生产环境为例)，设置 TCP 接入域名，进入 MQ 控制台的消费者管理页面，在左侧操作栏单击获取接入点获取
    	    properties.setProperty(PropertyKeyConst.ONSAddr, ONSAddr);
    	    producer = ONSFactory.createProducer(properties);
    	    // 在发送消息前，初始化调用start方法来启动Producer，只需调用一次即可，当项目关闭时，自动shutdown
    	    producer.start();
    	  }
    	 
    	  /**
    	   * 初始化生产者
    	   * @return
    	   */
    	  public Producer getProducer(){
    	    return producer;
    	  }
    }

**4. 生产者使用，生产推送消息：**

package com.sixmonth.rocketmq.service.rocketmq;
    
    import java.util.Date;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Service;
    
    import com.aliyun.openservices.ons.api.Message;
    import com.aliyun.openservices.ons.api.OnExceptionContext;
    import com.aliyun.openservices.ons.api.SendCallback;
    import com.aliyun.openservices.ons.api.SendResult;
    import com.sixmonth.rocketmq.common.rocketmq.init.RocketmqProducerInit;
    
    /**
     * 消息生产者，可与消费者分离；
     * 消息生产的三种方式：可靠同步发送，可靠异步发送，单向发送
     * @author sixmonth
     * @Date 2019年1月24日
     *
     */
    @Service
    public class RocketmqProducerService {
    	
    	private Logger logger = LoggerFactory.getLogger(RocketmqProducerService.class);
    	
    	@Value("${rocketmq.producer.msgTopic}")
    	private String msgTopic;
    	
    	@Autowired
    	private RocketmqProducerInit rocketmqProducerInit;
    	
    	public String tag = "*";//生产标签，可自定义，默认通配
    	
       /**
    	    * 同步发送实体对象消息
    	    * 可靠同步发送：同步发送是指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式；
    	    * 特点：速度快；有结果反馈；数据可靠；
    	    * 应用场景：应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等；
        */
       public boolean sendMsg(String msg) {
    	   Long startTime = System.currentTimeMillis();
    	   Message message = new Message(msgTopic, tag, msg.getBytes());
    	   SendResult sendResult = rocketmqProducerInit.getProducer().send(message);
    	   if (sendResult != null) {
               System.out.println(new Date() + " Send mq message success. Topic is:" + message.getTopic() + " msgId is: " + sendResult.getMessageId());
           } else {
    	     logger.warn(".sendResult is null.........");
    	   }
           Long endTime = System.currentTimeMillis();
    	   System.out.println("单次生产耗时："+(endTime-startTime)/1000);
    	   return true;
       }
       
       /**
        * 异步发送消息
        * 可靠异步发送：发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式；
        * 特点：速度快；有结果反馈；数据可靠；
        * 应用场景：异步发送一般用于链路耗时较长,对 rt响应时间较为敏感的业务场景,例如用户视频上传后通知启动转码服务,转码完成后通知推送转码结果等；
        * @param msg
        * @return
        */
       public boolean sendMsgAsy(String msg) {
    	   Long startTime = System.currentTimeMillis();
    	   Message message = new Message(msgTopic, tag, msg.getBytes());
    	   rocketmqProducerInit.getProducer().sendAsync(message, new SendCallback() {
    			@Override
    			public void onSuccess(SendResult sendResult) {
    				///消息发送成功
    	            System.out.println("send message success. topic=" + sendResult.getMessageId());
    			}
    	
    			@Override
    			public void onException(OnExceptionContext context) {
    				//消息发送失败
    				System.out.println("send message failed. execption=" + context.getException());
    			}
    	   });
           Long endTime = System.currentTimeMillis();
    	   System.out.println("单次生产耗时："+(endTime-startTime)/1000);
    	   return true;
       }
       
       /**
        * 单向发送
        * 单向发送：只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答；此方式发送消息的过程耗时非常短，一般在微秒级别；
        * 特点：速度最快，耗时非常短，毫秒级别；无结果反馈；数据不可靠，可能会丢失；
        * 应用场景：适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集；
        * @return
        */
       public boolean sendMsgOneway(String msg) {
    	   Long startTime = System.currentTimeMillis();
    	   Message message = new Message(msgTopic, tag, msg.getBytes());
    	   rocketmqProducerInit.getProducer().sendOneway(message);
           Long endTime = System.currentTimeMillis();
    	   System.out.println("单次生产耗时："+(endTime-startTime)/1000);
    	   return true;
       }
       
       
    }

*  **创建消费者，消费消息**（生产者和消费者可合并成一个项目，这里不做合并）

1. 引入依赖包：

2. yml配置文件添加属性：

#配置rocketmq      
    rocketmq:
      consumer:
    	producerId: GroupId_Test #消费者id(旧版本是消费者id，新版本是groupid)，替换成自己的
    	msgTopic: Test #消费主题，替换成自己的
    	accessKey: XXX  #连接通道，替换成自己的
    	secretKey: XXX  #连接秘钥，替换成自己的
    	onsAddr: http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet  #生产者ons接入域名，替换成自己的

**3. 初始化消费者：**

package com.sixmonth.rocketmq.common.rocketmq.init;
    
    import java.util.Properties;
    
    import javax.annotation.PostConstruct;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    import com.aliyun.openservices.ons.api.Consumer;
    import com.aliyun.openservices.ons.api.ONSFactory;
    import com.aliyun.openservices.ons.api.PropertyKeyConst;
    
    /**
     * rocketmq消费者启动初始化类
     * 同一个消费者可以消费不同的topic，同一个topic可以被不同的消费者消费（启用该方式相当于集群消费中的广播消费）
     * @author hqc
     * @Date 2019年1月17日
     *
     */
    @Component
    public class RocketmqConsumerInit {
    	
    	private Logger logger = LoggerFactory.getLogger(RocketmqConsumerInit.class);
    	
    	  @Value("${rocketmq.consumer.consumerId}")
    	  private String consumerId;
    	 
    	  @Value("${rocketmq.consumer.accessKey}")
    	  private String accessKey;
    	 
    	  @Value("${rocketmq.consumer.secretKey}")
    	  private String secretKey;
    	 
    	  @Value("${rocketmq.consumer.onsAddr}")
    	  private String ONSAddr;
    	  
    	  @Value("${rocketmq.consumer.msgTopic}")
    	  private String msgTopic;
    	  
    	  public static final String tag = "*";
    	 
    	  private static Consumer consumer;
    	  
    	  //public static final String topic2 = "Test2";//测试第二个消费主题
    	 
    	 
    	  @PostConstruct
    	  public void init(){
    		System.out.println("初始化启动消费者者！");
    	    // consumer 实例配置初始化
    	    Properties properties = new Properties();
    	    //您在控制台创建的consumer ID
    	    properties.setProperty(PropertyKeyConst.GROUP_ID, consumerId);
    	    // AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
    	    properties.setProperty(PropertyKeyConst.AccessKey, accessKey);
    	    // SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
    	    properties.setProperty(PropertyKeyConst.SecretKey, secretKey);
    	    //设置发送超时时间，单位毫秒
    	    properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
    	    // 设置 TCP 接入域名(此处以公共云生产环境为例)，设置 TCP 接入域名，进入 MQ 控制台的消费者管理页面，在左侧操作栏单击获取接入点获取
    	    properties.setProperty(PropertyKeyConst.ONSAddr, ONSAddr);
    	    consumer = ONSFactory.createConsumer(properties);
    	    consumer.subscribe(msgTopic, tag, new RocketmqTest1Listener());//监听第一个topic，new对应的监听器
    //	    consumer.subscribe(topic2, tag, new RocketmqTest2Listener());//监听另外一个topic，new对应的监听器
    	  // 在发送消息前，必须调用start方法来启动consumer，只需调用一次即可，当项目关闭时，自动shutdown
    	    consumer.start();
    	    logger.info("ConsumerConfig start success.");
    	    
    	  }
    	 
    	  /**
    	   * 初始化消费者
    	   * @return
    	   */
    	  public Consumer getconsumer(){
    	    return consumer;
    	  }
    }

**4. 消费者使用，消费消息**

package com.sixmonth.rocketmq.common.rocketmq.init;
    
    import java.util.ArrayList;
    import java.util.List;
    
    import javax.annotation.Resource;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Service;
    
    import com.aliyun.openservices.ons.api.Action;
    import com.aliyun.openservices.ons.api.ConsumeContext;
    import com.aliyun.openservices.ons.api.Message;
    import com.aliyun.openservices.ons.api.MessageListener;
    import com.sixmonth.rocketmq.common.util.Tool;
    
    /**
     * 监听消费
     * @author sixmonth
     * @Date 2019年1月17日
     *
     */
    @Service
    public class RocketmqTest1Listener implements MessageListener {
    
    	private Logger logger = LoggerFactory.getLogger(RocketmqTest1Listener.class);
    	
    	/**
    	 * 由于消费者是多线程的，所以对象要用static+set注入，把对象的级别提升到进程，
    	 * 这样多个线程就可以共用，但是无法调用父类的方法和变量
    	 */
    //	protected static TestDao testDao;
    	
    //    @Resource
    //    public void setTestDao(TestDao testDao){
    //    	RocketmqTest1Listener.testDao=testDao;//加入持久层dao，可根据需求自行修改
    //    }
    	
    	@Override
    	public Action consume(Message message, ConsumeContext context) {
    		try {
    		   Long startTime = System.currentTimeMillis();
    		   
    		   byte[] body = message.getBody();
    		   String msg = new String(body);//获取到接收的消息，由于接收到的是byte数组，所以需要转换成字符串
    		   
    		   //TODO 业务逻辑，自行设计
    		   //testDao.insertDatas();//持久层，这里不再展述，自行补全
    		   
    		   Long endTime = System.currentTimeMillis();
    		   System.out.println("单次消费耗时："+(endTime-startTime)/1000);
    		} catch (Exception e) {
    			logger.error("MessageListener.consume error:" + e.getMessage(), e);
    		}
    
    		logger.info("MessageListener.Receive message");
    		//如果想测试消息重投的功能,可以将Action.CommitMessage 替换成Action.ReconsumeLater
    		return Action.CommitMessage;
    	}
    
    }

### 开发扩展 ###

1.  官网已经提供了生产者和消费者项目demo实例，包括spring配置、tcp连接等功能，可自行下载参考：[https://download.csdn.net/download/alan\_liuyue/10935393][https_download.csdn.net_download_alan_liuyue_10935393]
2.  阿里云rocketmq消息队列开发者文档：[https://help.aliyun.com/document\_detail/29532.html?spm=a2c4g.11186623.6.541.2af177d7Qp5gOE][https_help.aliyun.com_document_detail_29532.html_spm_a2c4g.11186623.6.541.2af177d7Qp5gOE]

### 结语 ###

1.  rocketmq属于国产的阿里巴巴的自主研发的相对优秀的消息队列框架，能够承受双十一的巨大吞吐量，证明它的还是很值得选择的；
2.  2019新版本的生产者id和消费者id都统称为groupid，更加方便开发者接入，同时将ONSAddr换成NAMESRV\_ADDR，配置有细微修改，可在开发的时候自行修改，这里不作改动；
3.  如果需要自行搭建rocketmq集群来进行消息生产以及消费，可参考另外一篇博客：
4.  实践是检验认识真理性的唯一标准，亲手试过，发现真的能成功!!！

[https_help.aliyun.com_document_detail_94521.html_spm_a2c4g.11186623.6.543.18a07871mTGzwC]: https://help.aliyun.com/document_detail/94521.html?spm=a2c4g.11186623.6.543.18a07871mTGzwC
[https_download.csdn.net_download_alan_liuyue_10935393]: https://download.csdn.net/download/alan_liuyue/10935393
[https_help.aliyun.com_document_detail_29532.html_spm_a2c4g.11186623.6.541.2af177d7Qp5gOE]: https://help.aliyun.com/document_detail/29532.html?spm=a2c4g.11186623.6.541.2af177d7Qp5gOE