SpringBoot整合RabbitMQ之典型应用场景实战一-低调大师

SpringBoot整合RabbitMQ之典型应用场景实战一

2018-11-20 406

实战前言
RabbitMQ 作为目前应用相当广泛的消息中间件，在企业级应用、微服务应用中充当着重要的角色。特别是在一些典型的应用场景以及业务模块中具有重要的作用，比如业务服务模块解耦、异步通信、高并发限流、超时业务、数据延迟处理等。

RabbitMQ 官网拜读
首先，让我们先拜读 RabbitMQ 官网的技术开发手册以及相关的 Features，感兴趣的朋友可以耐心的阅读其中的相关介绍，相信会有一定的收获，地址可见：www.rabbitmq.com/getstarted.…

在阅读该手册过程中，我们可以得知 RabbitMQ 其实核心就是围绕 “消息模型” 来展开的，其中就包括了组成消息模型的相关组件：生产者，消费者，队列，交换机，路由，消息等！而我们在实战应用中，实际上也是紧紧围绕着 “消息模型” 来展开撸码的！

下面，我就介绍一下这一消息模型的演变历程，当然，这一历程在 RabbitMQ 官网也是可以窥览得到的！

上面几个图就已经概述了几个要点，而且，这几个要点的含义可以说是字如其名！
生产者：发送消息的程序
消费者：监听接收消费消息的程序
消息：一串二进制数据流
队列：消息的暂存区/存储区
交换机：消息的中转站，用于接收分发消息。其中有 fanout、direct、topic、headers 四种
路由：相当于密钥/第三者，与交换机绑定即可路由消息到指定的队列！

正如上图所展示的消息模型的演变，接下来我们将以代码的形式实战各种典型的业务场景！

SpringBoot 整合 RabbitMQ 实战
工欲善其事，必先利其器。我们首先需要借助 IDEA 的 Spring Initializr 用 Maven 构建一个 SpringBoot 的项目，并引入 RabbitMQ、Mybatis、Log4j 等第三方框架的依赖。搭建完成之后，可以简单的写个 RabbitMQController 测试一下项目是否搭建是否成功（可以暂时用单模块方式构建）

紧接着，我们进入实战的核心阶段，在项目或者服务中使用 RabbitMQ，其实无非是有几个核心要点要牢牢把握住，这几个核心要点在撸码过程中需要“时刻的游荡在自己的脑海里”，其中包括：
我要发送的消息是什么

我应该需要创建什么样的消息模型：DirectExchange+RoutingKey？TopicExchange+RoutingKey？等
我要处理的消息是实时的还是需要延时/延迟的？
消息的生产者需要在哪里写，消息的监听消费者需要在哪里写，各自的处理逻辑是啥

基于这样的几个要点，我们先小试牛刀一番，采用 RabbitMQ 实战异步写日志与异步发邮件。当然啦，在进行实战前，我们需要安装好 RabbitMQ 及其后端控制台应用，并在项目中配置一下 RabbitMQ 的相关参数以及相关 Bean 组件。

1.RabbitMQ 安装完成后，打开后端控制台应用：http://localhost:15672/ guest guest 登录，看到下图即表示安装成功

2.然后是项目配置文件层面的配置 application.properties

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.listener.concurrency=10
spring.rabbitmq.listener.max-concurrency=20
spring.rabbitmq.listener.prefetch=5

其中，后面三个参数主要是用于“并发量的配置”，表示：并发消费者的初始化值，并发消费者的最大值，每个消费者每次监听时可拉取处理的消息数量。

接下来，我们需要以 Configuration 的方式配置 RabbitMQ 并以 Bean 的方式显示注入 RabbitMQ 在发送接收处理消息时相关 Bean 组件配置其中典型的配置是 RabbitTemplate 以及 SimpleRabbitListenerContainerFactory，前者是充当消息的发送组件，后者是用于管理RabbitMQ监听器的容器工厂，其代码如下：

@Configuration
    public class RabbitmqConfig {
    private static final Logger log= LoggerFactory.getLogger(RabbitmqConfig.class);
    @Autowired
    private Environment env;
    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;
    /**
     * 单一消费者
     * @return
     */
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        factory.setTxSize(1);
        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        return factory;
    }
    /**
     * 多个消费者
     * @return
     */
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory,connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        factory.setConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.concurrency",int.class));
        factory.setMaxConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.max-concurrency",int.class));
        factory.setPrefetchCount(env.getProperty("spring.rabbitmq.listener.prefetch",int.class));
        return factory;
    }
    @Bean
    public RabbitTemplate rabbitTemplate(){
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                log.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
            }
        });
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);
            }
        });
        return rabbitTemplate;
    }}

RabbitMQ 实战：业务模块解耦以及异步通信
在一些企业级系统中，我们经常可以见到一个执行 function 通常是由许多子模块组成的，这个 function 在执行过程中，需要同步的将其代码从头开始执行到尾，即执行流程是 module_A -> module_B -> module_C -> module_D，典型的案例可以参见汇编或者 C 语言等面向过程语言开发的应用，现在的一些 JavaWeb 应用也存在着这样的写法。

而我们知道，这个执行流程其实对于整个 function 来讲是有一定的弊端的，主要有两点：
整个 function 的执行响应时间将很久；
如果某个 module 发生异常而没有处理得当，可能会影响其他 module 甚至整个 function 的执行流程与结果；

整个 function 中代码可能会很冗长，模块与模块之间可能需要进行强通信以及数据的交互，出现问题时难以定位与维护，甚至会陷入 “改一处代码而动全身”的尴尬境地！

故而，我们需要想办法进行优化，我们需要将强关联的业务模块解耦以及某些模块之间实行异步通信！下面就以两个场景来实战我们的优化措施！

场景一：异步记录用户操作日志
对于企业级应用系统或者微服务应用中，我们经常需要追溯跟踪记录用户的操作日志，而这部分的业务在某种程度上是不应该跟主业务模块耦合在一起的，故而我们需要将其单独抽出并以异步的方式与主模块进行异步通信交互数据。

下面我们就用 RabbitMQ 的 DirectExchange+RoutingKey 消息模型也实现“用户登录成功记录日志”的场景。如前面所言，我们需要在脑海里回荡着几个要点：
消息模型：DirectExchange+RoutingKey 消息模型
消息：用户登录的实体信息，包括用户名，登录事件，来源的IP，所属日志模块等信息
发送接收：在登录的 Controller 中实现发送，在某个 listener 中实现接收并将监听消费到的消息入数据表；实时发送接收

首先我们需要在上面的 RabbitmqConfig 类中创建消息模型：包括 Queue、Exchange、RoutingKey 等的建立，代码如下：

上图中 env 获取的信息，我们需要在 application.properties 进行配置，其中 mq.env=local

此时，我们将整个项目/服务跑起来，并打开 RabbitMQ 后端控制台应用，即可看到队列以及交换机及其绑定已经建立好了，如下所示：

接下来，我们需要在 Controller 中执行用户登录逻辑，记录用户登录日志，查询获取用户角色视野资源信息等，由于篇幅关系，在这里我们重点要实现的是用MQ实现 “异步记录用户登录日志” 的逻辑，即在这里 Controller 将充当“生产者”的角色，核心代码如下：

@RestController
    public class UserController {
    private static final Logger log= LoggerFactory.getLogger(HelloWorldController.class);
    private static final String Prefix="user";
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private UserMapper userMapper;
    @Autowired
    private UserLogMapper userLogMapper;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private Environment env;
    @RequestMapping(value = Prefix+"/login",method = RequestMethod.POST,consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public BaseResponse login(@RequestParam("userName") String userName,@RequestParam("password") String password){
        BaseResponse response=new BaseResponse(StatusCode.Success);
        try {
            //TODO：执行登录逻辑
            User user=userMapper.selectByUserNamePassword(userName,password);
            if (user!=null){
                //TODO：异步写用户日志
                try {
                    UserLog userLog=new UserLog(userName,"Login","login",objectMapper.writeValueAsString(user));
                    userLog.setCreateTime(new Date());
                    rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
                    rabbitTemplate.setExchange(env.getProperty("log.user.exchange.name"));
                    rabbitTemplate.setRoutingKey(env.getProperty("log.user.routing.key.name"));
                    Message message=MessageBuilder.withBody(objectMapper.writeValueAsBytes(userLog)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
                    message.getMessageProperties().setHeader(AbstractJavaTypeMapper.DEFAULT_CONTENT_CLASSID_FIELD_NAME, MessageProperties.CONTENT_TYPE_JSON); 
                    rabbitTemplate.convertAndSend(message);         
                }catch (Exception e){
                    e.printStackTrace();
                }
                //TODO：塞权限数据-资源数据-视野数据
            }else{
                response=new BaseResponse(StatusCode.Fail);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return response;
    }}

在上面的“发送逻辑”代码中，其实也体现了我们最开始介绍的演进中的几种消息模型，比如我们是将消息发送到 Exchange 的而不是 Queue，消息是以二进制流的形式进行传输等等。当用 postman 请求到这个 controller 的方法时，我们可以在 RabbitMQ 的后端控制台应用看到一条未确认的消息，通过 GetMessage 即可看到其中的详情，如下：

最后，我们将开发消费端的业务代码，如下：

@Component
    public class CommonMqListener {
    private static final Logger log= LoggerFactory.getLogger(CommonMqListener.class);
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private UserLogMapper userLogMapper;
    @Autowired
    private MailService mailService;
    /**
     * 监听消费用户日志
     * @param message
     */
    @RabbitListener(queues = "${log.user.queue.name}",containerFactory = "singleListenerContainer")
    public void consumeUserLogQueue(@Payload byte[] message){
        try {
            UserLog userLog=objectMapper.readValue(message, UserLog.class);
            log.info("监听消费用户日志 监听到消息： {} ",userLog);
            //TODO：记录日志入数据表
            userLogMapper.insertSelective(userLog);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

将服务跑起来之后，我们即可监听消费到上面 Queue 中的消息，即当前用户登录的信息，而且，我们也可以看到“记录用户登录日志”的逻辑是由一条异于主业务线程的异步线程去执行的：

“异步记录用户操作日志”的案例我想足以用于诠释上面所讲的相关理论知识点了，在后续篇章中，由于篇幅限制，我将重点介绍其核心的业务逻辑！

场景二：异步发送邮件
发送邮件的场景，其实也是比较常见的，比如用户注册需要邮箱验证，用户异地登录发送邮件通知等等，在这里我以 RabbitMQ 实现异步发送邮件。实现的步骤跟场景一几乎一致！

消息模型的创建
配置信息的创建
生产端
消费端

彩蛋：本博文就先介绍RabbitMQ实战的典型业务场景之业务服务模块异步解耦与通信吧，下篇博文将继续讲解RabbitMQ实战在高并发系统的场景的应用记忆消息确认机制跟并发量的配置实战！另外，博主已将RabbitMQ相关技术以及场景实战的相关要点录制成了视频教程，感兴趣小伙伴可以前往学习观看：RabbitMQ实战！！

微信关注我们

原文链接：https://yq.aliyun.com/articles/672235

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

（二）spring cloud微服务分布式云架构 - 整合企业架构的技术点

spring cloud本身提供的组件就很多，但我们需要按照企业的业务模式来定制企业所需要的通用架构，那我们现在需要考虑使用哪些技术呢？下面我针对于spring cloud微服务分布式云架构做了以下技术总结，希望可以帮助到大家： View：H5、Vue.js、Spring Tag、React、angularJs Spring Boot/Spring Cloud：Zuul、Ribbon、Feign、Turbine、Hystrix、Oauthor2、Sleuth、API Gateway、Spring Cloud、Config Eureka、SSO、Spring Cloud、BUS、Turbine、Zipkin、Cache、Spring Cloud Admin、API Gateway、ELK Spring Cloud Security、 Spring Cloud Stream Component：RoketMQ、Kafka、MongoDB、OSS、Redis、Swagger、Zuul、Label、BASE、Charts、Utils DAO：Spring Data、Mybatis、OSS、 ...

2018-11-21

510

Spring Cloud集成项目有很多，下面我们列举一下和Spring Cloud相关的优秀项目，我们的企业架构中用到了很多的优秀项目，说白了，也是站在巨人的肩膀上去整合的。在学习Spring Cloud之前大家必须了解一下相关项目，希望可以帮助到大家。 Spring Cloud Config 配置管理工具包，让你可以把配置放到远程服务器，集中化管理集群配置，目前支持本地存储、Git以及Subversion。 Spring Cloud Bus 事件、消息总线，用于在集群（例如，配置变化事件）中传播状态变化，可与Spring Cloud Config联合实现热部署。 Eureka 云端服务发现，一个基于 REST 的服务，用于定位服务，以实现云端中间层服务发现和故障转移。 Hystrix 熔断器，容错管理工具，旨在通过熔断机制控制服务和第三方库的节点,从而对延迟和故障提供更强大的容错能力。 Zuul Zuul 是在云平台上提供动态路由,监控,弹性,安全等边缘服务的框架。Zuul 相当于是设备和 Netflix 流应用的 Web 网站后端所有请求的前门。 Archaius 配置管理API...

2018-11-21

564

资源下载

更多资源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。