SpringBoot从入门到精通教程（三）- RocketMQ集成和场景案例分析

梦里梦外; 2022-01-29 07:27 531阅读 0赞

## 需求背景 ##

> RocketMq是一个由阿里巴巴开源的消息中间件，在设计上借鉴了Kafka，其2012年开源，2017年成为Apache顶级项目。
> 
> 一般MQ适用场景如下：
> 
> 1. 流量削峰，提升系统高并发处理能力，比如秒杀场景  
> 2. 异步处理，应用解耦，提高系统吞吐量
> 
> RocketMQ基础组件：
> 
> 1. Producer是消息生产者  
> 2. Consumer是消息消费者  
> 3. Topic是消息传递中间者，其中存放的是消息逻辑地址

## 技术点 ##

主要使用了两个

1. RocketMQTemplate对象（com.zc.smartcity.rocketmq.core.RocketMQTemplate）

2. spring-boot-starter-rocketmq组件

注：第2点，通过maven官方中央仓库[https://search.maven.org/][https_search.maven.org]，搜索到了几个，我使用了最新日期的，即第一个

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlbWluMTAwMw_size_16_color_FFFFFF_t_70][]

## 案例说明 ##

场景案例分析：

> 1. 如何与springboot集成  
> 2. 如何建立连接，发送不同消息数据类型  
> 3. 如何订阅，且消费不同消息数据类型  
> 4. 场景案例分析

## 代码演示 ##

举两个例子

**1. 如何与springboot集成**

pom.xml文件

<dependency>
      <groupId>com.github.org-hejianhui</groupId>
      <artifactId>spring-boot-starter-rocketmq</artifactId>
      <version>1.0-RELEASE</version>
    </dependency>

application.yml文件

spring:
      # mq service
      rocketmq: 
        name-server: localhost:9876
        producer: 
          group: my-mq-group
          send-message-timeout: 86400
          compress-message-body-threshold: 4096
          max-message-size: 4194304
          retry-times-when-send-async-failed: 0
          retry-next-server: true
          retry-times-when-send-failed: 2

注：这些参数含义，其英文名称理解起来应该不难，就不做进一步解释了

**2. 如何建立连接，发送不同消息数据类型**

package com.md.demo.rest;
    
    import java.math.BigDecimal;
    import java.util.Random;
    
    import javax.annotation.Resource;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import com.md.demo.MyTopics;
    import com.md.demo.vo.OrderPaidEvent;
    import com.zc.smartcity.rocketmq.core.RocketMQTemplate;
    
    /**
     * @author Minbo
     */
    @RestController
    public class InitRest {
    
    	protected static Logger logger = LoggerFactory.getLogger(InitRest.class);
    
    	@Resource
    	private RocketMQTemplate rocketMQTemplate;
    
    	/**
    	 * 发送普通文本消息
    	 * http://localhost:9090/sendMsgText
    	 * @return
    	 */
    	@RequestMapping("/sendMsgText")
    	public String sendMsgText() {
    		// 发送普通文本消息
    		String msg = "Hello，World，spring-boot2-rocketmq! flag=" + new Random().nextInt(100);
    		String destination1 = MyTopics.TOPIC1 + ":" + MyTopics.TAG1;
    		rocketMQTemplate.convertAndSend(destination1, msg);
    		logger.info("sendMsg success。发送普通文本消息 msg=" + msg);
    
    		return "success for text";
    	}
    
    	/**
    	 * 发送对象数据消息
    	 * http://localhost:9090/sendMsgObj
    	 * @return
    	 */
    	@RequestMapping("/sendMsgObj")
    	public String sendMsgObj() {
    		// 发送对象数据消息
    		OrderPaidEvent order = new OrderPaidEvent("T_001_ID_" + new Random().nextInt(100), new BigDecimal("88.00"));
    		String destination2 = MyTopics.TOPIC1 + ":" + MyTopics.TAG2;
    		rocketMQTemplate.convertAndSend(destination2, order);
    		logger.info("sendMsg success。发送对象消息 order=" + order.toString());
    
    		return "success for obj";
    	}
    
    	/*
    	 * RocketMQ的最佳实践中推荐：一个应用尽可能用一个Topic，消息子类型用tags来标识，tags可以由应用自由设置。
    	 * 在使用rocketMQTemplate发送消息时，通过设置发送方法的destination参数来设置消息的目的地，
    	 * destination的格式为topicName:tagName，:前面表示topic的名称，后面表示tags名称。
    	 * 
    	 * 注意: tags从命名来看像是一个复数，但发送消息时，目的地只能指定一个topic下的一个tag，不能指定多个。
    	 */
    }

## 完整源码下载 ##

我的[Github源码地址：][Github]

[https://github.com/hemin1003/spring-boot-study/tree/master/spring-boot2-study/spring-boot2-parent][Github]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlbWluMTAwMw_size_16_color_FFFFFF_t_70 1][]

## 附加说明 ##

**使用命令新建Topic**

1、安装完rocketmq后（[官方安装教程][Link 1]），进入到apache-rocketmq目录  
2、使用命令

sh /bin/mqadmin updateTopic -n localhost:9876 -b localhost:10911 -t test-topic-1

3、如图演示

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlbWluMTAwMw_size_16_color_FFFFFF_t_70 2][]

**updateTopic命令主要是三个参数：**

\-n nameserver地址和端口，一般启动broker时要指定  
\-b broker\_ip:broker\_port，ip是启动broker机器的ip，默认是10911  
\-t 新主题名称

**问题：注意事项**（autoCreateTopicEnable=true）

1、使用默认设置autoCreateTopicEnable=true，这样会导致topic的设置不能规范管理，没有统一的审核等  
2、使用默认设置autoCreateTopicEnable=true，还会造成负载均衡失效

所以，一般生产环境中会设置autoCreateTopicEnable=false

设置方式：启动broker时关闭掉，命令如下

nohup sh bin/mqbroker -n localhost:9876 autoCreateTopicEnable=false &

## 常见错误 ##

**1、call timeout 超时异常**

org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout] with root cause
    org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout
    	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:635)
    	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1280)
    	at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:301)
    	at com.zc.smartcity.rocketmq.core.RocketMQTemplate.syncSend(RocketMQTemplate.java:126)
    	at com.zc.smartcity.rocketmq.core.RocketMQTemplate.syncSend(RocketMQTemplate.java:101)
    	at com.zc.smartcity.rocketmq.core.RocketMQTemplate.syncSend(RocketMQTemplate.java:89)
    	at com.zc.smartcity.rocketmq.core.RocketMQTemplate.doSend(RocketMQTemplate.java:437)
    	at com.zc.smartcity.rocketmq.core.RocketMQTemplate.doSend(RocketMQTemplate.java:28)

解决办法：调大send-message-timeout值（application.yml）

**2、No route info of this topic 异常**

See http://rocketmq.apache.org/docs/faq/ for further details.; nested exception is org.apache.rocketmq.client.exception.MQClientException: No route info of this topic, test-topic-1
    See http://rocketmq.apache.org/docs/faq/ for further details.] with root cause
    org.apache.rocketmq.client.exception.MQClientException: No route info of this topic, test-topic-1
    See http://rocketmq.apache.org/docs/faq/ for further details.
    	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:657)
    	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1280)
    	at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:301)
    	at com.zc.smartcity.rocketmq.core.RocketMQTemplate.syncSend(RocketMQTemplate.java:126)
    	at com.zc.smartcity.rocketmq.core.RocketMQTemplate.syncSend(RocketMQTemplate.java:101)
    	at com.zc.smartcity.rocketmq.core.RocketMQTemplate.syncSend(RocketMQTemplate.java:89)
    	at com.zc.smartcity.rocketmq.core.RocketMQTemplate.doSend(RocketMQTemplate.java:437)
    	at com.zc.smartcity.rocketmq.core.RocketMQTemplate.doSend(RocketMQTemplate.java:28)

解决方法：参考以上《附加说明》的步骤，需要先新建topic后才能发送消息或消费消息。这里不再做进一步解释

## 其他官方资料 ##

1、[RocketMq官方网址][RocketMq]

2、[RocketMq官方安装详细步骤，和案例演示说明][Link 1]

3、[RocketMQ-SpringBoot官方代码Demo][RocketMQ-SpringBoot_Demo]

4、[RocketMQ-SpringBoot官方代码Demo-中文教程说明][RocketMQ-SpringBoot_Demo-]

5、[RocketMQ控制台-运维管理系统搭建文档][RocketMQ_-]

6、[RocketMQ-常见问题FAQ][RocketMQ-SpringBoot_Demo-]

## 下一章教程 ##

[SpringBoot从入门到精通教程（四）- @Scheduled定时器用法和场景案例分析][SpringBoot_- _Scheduled]

## 该系列教程 ##

[SpringBoot从入门到精通教程][SpringBoot]

\------------------------------------------------------

[关于我][Link 2]（个人域名）

[我的开源项目集Github][Github 1]

期望和大家一起学习，共同进步，共勉，O(∩\_∩)O谢谢

欢迎交流问题，可加个人QQ 469580884，

或者，加我的群号 **751925591**，一起探讨交流问题

不讲虚的，只做实干家

Talk is cheap，show me the code

![20180411181545874][]

如果觉得内容赞，您可以请我喝杯咖啡：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9oZW1pbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70][]        ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9oZW1pbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1][]

[https_search.maven.org]: https://search.maven.org/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlbWluMTAwMw_size_16_color_FFFFFF_t_70]: /images/20220129/f396d7a98b364c74a988b2b564cd3886.png
[Github]: https://github.com/hemin1003/spring-boot-study/tree/master/spring-boot2-study/spring-boot2-parent
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlbWluMTAwMw_size_16_color_FFFFFF_t_70 1]: /images/20220129/119194580c9b4eda89d46574f6295f21.png
[Link 1]: http://rocketmq.apache.org/docs/quick-start/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlbWluMTAwMw_size_16_color_FFFFFF_t_70 2]: /images/20220129/a27009a620c44fc7b1a5322988cb8110.png
[RocketMq]: http://rocketmq.apache.org/
[RocketMQ-SpringBoot_Demo]: https://github.com/hemin1003/rocketmq-spring
[RocketMQ-SpringBoot_Demo-]: https://github.com/hemin1003/rocketmq-spring/blob/master/README_zh_CN.md
[RocketMQ_-]: https://github.com/apache/rocketmq-externals/blob/master/rocketmq-console/doc/1_0_0/UserGuide_CN.md
[SpringBoot_- _Scheduled]: https://blog.csdn.net/hemin1003/article/details/90454462
[SpringBoot]: https://hemin.blog.csdn.net/column/info/40170
[Link 2]: http://heminit.com/about/
[Github 1]: https://github.com/hemin1003
[20180411181545874]: /images/20220129/f7900febec3c4b0e9c2230aabf544e89.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9oZW1pbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190816110426562.jpeg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9oZW1pbi5ibG9nLmNzZG4ubmV0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9oZW1pbi5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20190816110923699.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9oZW1pbi5ibG9nLmNzZG4ubmV0,size_16,color_FFFFFF,t_70