—— 基于点对点(自动 Ack)
SpringBoot 集成 RabbitMQ,如果只是简单的使用,配置非常少,SpringBoot 提供了 spring-boot-starter-amqp 项目对消息各种支持。本文以简单版(一个生产者和一个消费者)的形式入门,使用 默认的消息交换机(Direct Exchange)
[TOC]
消息中间件
消息队列中间件
指利用高效可靠的消息传递机制进行与平台无关的数据交流,并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型,它可以在分布式环境下扩展进程间的通信。 消息队列中间件
也称为 消息队列
和 消息中间件
。它一般有两种传递模式:点对点模式 和 发布/订阅模式。
消息中间件的作用
解耦
:消息中间件在处理过程中插入了一个隐含的、基于数据的接口层,两边的处理过程都要实现这一层接口,这允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束即可。存储
:在某些情况下,处理数据的过程会失败。消息中间件可以把数据进行持久化处理直到它们已经被完全处理,通过这一方式规避了数据丢失的风险。在把一个消息从消息中间件中删除之前,需要你的处理系统明确指出该消息已经被处理完成,从而确保你的数据被安全的保存直到你使用完毕。削峰
:在访问量剧增的情况下,应用仍然需要继续发挥作用,但是这样的突发流量并不常见。如果以处理这类峰值为标准投入资源,无疑是巨大的浪费。使用消息中间件能够使用关键组件支撑突发访问压力,不会因为突发的超负荷请求而完全崩溃。可恢复性
:当系统的一部分组件失效时,不会影响到整个系统。消息中间件降低了进程间的耦合度,所以即使一个处理消息的进程挂掉,加入消息中间件的消息仍然可以在系统恢复后进行处理。顺序保证
:在大多数使用场景下,数据处理的顺序很重要,大部分消息中间件支持一定程度上的顺序性。缓冲
:在任何重要的系统中,都会存在需要不同处理时间的元素。消息中间件通过一个缓冲层来帮助任务最高效率的执行,写入消息中间件的处理会尽可能快速。该缓冲层有助于控制和优化数据流经过系统的速度。异步通性
:在很多时候应用不想也不需要立即处理消息。消息中间件提供了异步处理机制,允许应用把一些消息放入消息中间件中,但不立即处理它,在之后需要的时候再慢慢处理。
RabbitMQ介绍
AMQP,即Advanced Message Queuing Protocol,高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。消息中间件主要用于组件之间的解耦,消息的发送者无需知道消息使用者的存在,反之亦然。AMQP 的主要特征是 面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。
RabbitMQ 是实现 AMQP(高级消息队列协议)的消息中间件的一种,最初起源于金融系统,用于在分布式系统中存储转发消息,在 易用性、扩展性、高可用性
等方面表现不俗。RabbitMQ 服务器端用 Erlang 语言 编写,支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等,支持AJAX。支持延迟队列(这是一个非常有用的功能)
….
JMS 是指一套规范,它是有 Sun 公司提出来的,实际就是指 JMS API,旨在为 java 应用提供一套统一的 API(类似 JPA 吧)。AMQP 则是一种协议,RabbitMQ 则是基于这种协议产生的消息中间件。RabbitMQ 还能作为 JMS 的实现,因为它支持 JMS 这种客户端(类似 Spring Cache,可以有很多实现,其中之一就是 Spring Data Redis,前面的整合笔记用过呢)。
环境/版本一览:
- 开发工具:Intellij IDEA 2018.2.2
- springboot: 2.0.6.RELEASE
- jdk:1.8.0_171
- maven:3.3.9
- spring-boot-starter-amqp: 2.0.6.RELEASE
1、pom.xml
1 | <dependencies> |
2、application.yml
1 | spring: |
3、config
1 | package com.fatal.config; |
4、sender
1 | package com.fatal.sender; |
5、receiver
1 | package com.fatal.receiver; |
6、controller
1 | package com.fatal.controller; |
7、测试
7.1、测试抛正常的 autoAck
注掉FatalReceiver
类 22 行的int i = 1/0;
的注释,然后运行项目
RabbitMQ 客户端:
控制台:
7.2、测试抛异常的 autoAck
打开FatalReceiver
类 22 行的int i = 1/0;
,然后运行项目
你会发现循环异常,这时候终止程序。。。
RabbitMQ 客户端:
控制台:
这是简单的入门,使用自动 Ack 的方式。当然咯,缺陷如 7.2 的测试结果一样,所以不推荐使用 自动 Ack
笔记
基础概念
通常我们谈到队列服务, 会有三个概念: 发消息者、队列、收消息者,RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和 队列之间, 加入了交换器 (Exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。
- 左侧 P 代表 生产者,也就是往 RabbitMQ 发消息的程序。
- 中间即是 RabbitMQ Broker,其中包括了 交换机 和 队列。
- 右侧 C 代表 消费者,也就是往 RabbitMQ 拿消息的程序。
Broker:简单来说就是消息队列服务器实体
Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列
Queue:消息队列载体,每个消息都会被投入到一个或多个队列
Binding:绑定,它的作用就是把exchange
和queue
按照路由规则绑定起来,在绑定的时候一定会指定一个绑定键“Binding Key
”
Routing Key:路由关键字,exchange
根据这个关键字进行消息投递
vhost:虚拟主机,一个broker
里可以开设多个vhost
,用作不同用户的权限分离
producer:消息生产者,就是投递消息的程序
consumer:消息消费者,就是接受消息的程序
channel:消息通道,在客户端的每个连接里,可建立多个channel
,每个channel
代表一个会话任务
常见应用场景
- 邮箱发送:用户注册后投递消息到
rabbitmq
中,由消息的消费方异步的发送邮件,提升系统响应速度 - 流量削峰:一般在 秒杀活动 中应用广泛,秒杀会因为流量过大,导致应用挂掉,为了解决这个问题,一般在应用前端加入消息队列。用于控制活动人数,将超过此一定阀值的订单直接丢弃。缓解短时间的高流量压垮应用。
- 订单超时:利用
rabbitmq
的 延迟队列,可以很简单的实现 订单超时 的功能,比如用户在下单后30分钟未支付取消订单 - 还有更多应用场景就不一一列举了…..
怎么知道没有配置交换机的时候是使用Direct
作为默认交换机呢?
源码
点开 RabbitTemplate 的源码来看一下
它这里定义了一个常量 DEFAULT_EXCHANGE
然后,当我们使用 RabbitTemplate 发布消息的时候,如果我们没有指定交换机,那么调用的方法类似下面的 convertAndSend(String routingKey, final Object object)
,可以看出,它帮我们指定了默认的交换机了。
RabbitMQ 客户端
RabbitMQ 已经帮我们初始化了一些交换机,其中就包括默认交换机
点进去查看它的 Bindings,你可以看到一串描述
1 | The default exchange is implicitly bound to every queue, with a routing key equal to the queue name. It is not possible to explicitly bind to, or unbind from the default exchange. It also cannot be deleted. |
参考资料
一起来学SpringBoot | 第十二篇:初探RabbitMQ消息队列
RabbitMQ 实战指南
总结
SpringBoot
的知识已经有前辈在我们之前探索了。比较喜欢的博主有:唐亚峰 | Battcn、方志朋的专栏、程序猿DD、纯洁的微笑。对这门技术感兴趣的可以去他们的博客逛逛。谢谢他们的分享~~
以上文章是我用来学习的Demo
,都是基于 SpringBoot2.x
版本。
源码地址: https://github.com/ynfatal/springboot2-learning/tree/master/chapter19_1