RabbItMQ核心概念:
RabbitMQ是一个开源的消息代理和队列服务器,用来通过普通协议在完全不同的应用之间共享数据,RabbitMQ是用Erlang语言来编写的,并且RabbitMQ是基于AMQP协议的。
1、RabbitMQ的高性能之道是如何做到的?
Erlang语言最初在于交换机领域的架构模式,这样使得RabbitMQ在Broker之间进行数据交互的性能是非常优秀的。
Erlang的优点:Erlang有着和原生Socket一样的延迟。
2、什么是AMQP高级消息队列协议?
AMQP:是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。
AMQP核心概念:
-
Server:又称Broker,接受客户端的连接,实现AMQP实体服务
-
Connection:连接,应用程序与Broker的网络连接
-
Channel:网络信道,几乎所有的操作都在Channel中进行,Channel是进行消息读写的通道。客户端可建立多个Channel,每个Channel代表一个会话任务。
-
Message:消息,服务器和应用程序之间传送的数据,由Properties和Body组成。Properties可以对消息进行修饰,比如消息的优先级、延迟等高级特性;Body则就是消息体内容。
-
Virtual host:虚拟地址,用于进行逻辑隔离,最上层的消息路由。一个Virtual Host里面可以有若干个Exchange和Queue,同一个Virtual host里面不能有相同名称的Exchange和Queue。
-
Exchange:交换机,接收消息,根据路由键转发消息到绑定的队列
-
Binding:Exchange和Queue之间的虚拟连接,binding中可以包含Routing key
-
Routing Key :一个路由规则,虚拟机可以用它来确定如何路由一个特定消息。
-
Queue:也称为Message Queue,消息队列,保存消息并将它们转发给消费者
3、RabbitMQ消息是如何流转的?
4、Exchange交换机详解
接收消息,并根据路由键转发消息到所绑定的队列。
交换机属性:
-
Name:交换机名称
-
Type:交换机类型为direct、topic、fanout、headers
- Direct Exchange:所有发送到Direct Exchange的消息被转发到RouteKey中指定的Queue(不需要进行任何绑定操作)
- Topic Exchage:所有发送到Topic Exchange的消息被转发到所有关心RouteKey中指定Topic的Queue上,Exchange将RouteKey和某Topic进行模糊匹配,此时队列需要绑定一个Topic。
Fanout Exchange:
1、不处理路由键,只需要简单的将队列绑定到交换机上
2、发送到交换机的消息都会被转发到与该交换机绑定的所有队列上
3、Fanout交换机转发消息是最快的。(不做任何匹配处理)
Queue-消息队列
- 消息队列,实际存储消息数据
- Durability:是否持久化,Durable:是,Transient:否
- Auto delete:如选yes,代表当最后一个监听被移除后,该Queue会自动被删除
Virtual host-虚拟主机:
- 虚拟地址,用于进行逻辑隔离,最上层的消息路由
- 一个Virtual Host里面可以有若干个Exchange和Queue
- 同一个Virtual Host里面不能有相同名称的Exchange和Queue
RabbitMQ高级特性:
1、消息如何保障100%的投递成功?
-
消息落库,对消息状态进行打标
-
消息的延迟投递,做二次确认,回调检查
2、消费端—幂等性保障
-
唯一ID+指纹码机制
好处:实现简单
坏处:高并发下有数据库写入的性能瓶颈
解决方案:根据id进行分库分表进行算法路由
-
利用redis原子特性实现
3、Confirm确认消息
理解Comfirm消息确认机制:
- 消息的确认,是指生产者投递消息后,如果Broker收到消息,则会给我们生产者一个应答。
- 生产者进行接收应答,用来确定这条消息是否正常的发送到Broker,这种方式也是消息的可靠性投递的核心保障!
如何实现Confirm确认消息?
第一步:在channel上开启确认模式:channel.confirmSelect()
第二步:在channel上添加监听:addConfirmListener,监听成功和失败的 返回结果,根据具体的结果对消息进行重新发送、或记录日志等后续处理!
Return消息机制:Return Listener 用于处理一些不可路由的消息!
- 被 broker 接收到只能表示 message 已到达服务器,并不能保证消息一定会被投递到目标 queue。所以还需 returnCallback 。
- 我们的消息生产者,通过指定一个Exchange和RoutingKey,把消息送达到其一个队列中去,然后我们的消费者监听队列,进行消费处理操作!
- 但是在某些情况下,如果我们在发送消息的时候,当前的Exchange不存在或者指定的路由key路由不到,这个时候如果我们需要监听这种不可达的消息,就要使用Return Listener!
4、TTL队列、消息
- TTL是Time To Live的缩写,也就是生存时间
- RabbitMQ支持消息的过期时间,在消息发送时可以进行指定
- RabbitMQ支持队列的过期时间,从消息入队列开始计算,只要超过队列超时时间配置,那么消息会自动的清除
5、死信队列
利用DLX,当消息在一个队列中变成了死信之后,它能被重新publish到另一个Exchange,这个Exchange就是DLX。
消息变成死信有以下几种情况:
- 消息被拒绝,并且requeue=false
- 消息TTL过期
- 队列达到最大长度
6、RabbitMQ的使用场景:
1、跨系统的异步通信
2、多个应用之间的耦合
3、应用内的同步变异步
4、消息驱动的架构
5、应用需要更灵活的耦合方式
6、跨局域网
Springboot整合RabbitMQ
消息提供者:
先导入相对应的依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
然后编写消息提供者的配置文件
#对于rabbitMQ的支持
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
#项目的名称
spring.application.name=rabbitmq-provider
再编写消息提供者的配置类,写好消息的队列、交换机、绑定策略等
@Configuration
public class RabbitConfig {
@Bean
public Queue TestDirectQueue(){
return new Queue("TestDirectQueue1111",true);
}
@Bean
public DirectExchange TestDirectExchange(){
return new DirectExchange("TestDirectExchange",true,false);
}
@Bean
public Binding bindingDirect() {
return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
}
@Bean
DirectExchange lonelyDirectExchange() {
return new DirectExchange("lonelyDirectExchange");
}
}
最后在编写消息提供者的接口层
@RestController
public class SendMessageController {
@Autowired
RabbitTemplate rabbitTemplate;
@GetMapping("/sendDirectMessage")
public String sendDirectMessage() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "test message, hello!";
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String,Object> map=new HashMap<>();
map.put("messageId",messageId);
map.put("messageData",messageData);
map.put("createTime",createTime);
//将消息携带绑定键值:TestDirectRouting 发送到交换机TestDirectExchange
rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
return "ok";
}
}
启动项目,访问该接口,就能在RabbitMQ的可视化界面可以看到该队列已经存在。
RabbitMQ消费者:
一样的先导入依赖:
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
配置application.properties配置文件
server.port=8081
#对于rabbitMQ的支持
spring.rabbitmq.host=192.168.202.231
spring.rabbitmq.port=5672
spring.rabbitmq.username=oilchem
spring.rabbitmq.password=PoHF*fn%
#项目的名称
spring.application.name=rabbitmq-consumer
这时候其实消费者不需要再写配置类了,直接写相关业务
@Component
@RabbitListener(queues = "TestDirectQueue1111")
public class DirectReceiver {
@RabbitHandler
public void process(Map testMessage){
System.out.println("DirectReceiver消费者接收的消息为"+testMessage);
}
}
这时候启动消息消费者就能取到指定队列中的消息了
这样就实现了一个最简单的Springboot整合RabbitMQ的案例了。