rabbitMQ
一、应用场景
1.异步处理
普通业务
普通处理，每个阶段50ms，总共150ms
异步
写入数据库后，异步处理，响应时间为最高的50ms,所以总共响应100ms
消息队列
当注册成功，写入数据库后，将成功的消息写入消息队列，写入时间耗时非常短，然后直接返回给用户。其他服务，例如邮件发送、短信发送的服务通过消息队列拿到注册成功的消息后，再进行发送。通俗的来说，我们不关心邮件和短信发送成功与否，注册成功了就能使用功能。毕竟没有人看到页面提示注册成功，而憨憨的等着邮件通知后再使用。
2、解耦
普通方式，直接请求，但是万一库存服务崩溃了，总不可能让用户自己怀疑网络的原因而下单失败吧。
解耦后：
使用消息队列后，用户下单后，将成功的消息写入消息队列，库存订阅消息，拿到消息后进行减库存操作，崩溃了虽然拿不到，但是用户知道自己下单成功了，库存恢复了照样拿，然后减库存。
3.流量削峰
如果请求直接进来，太多了会受不了的，然后崩溃了咋办。所以利用消息队列，全部写入消息队列，然后后端慢慢处理。如果硬要说消息队列里太多了，塞不下咋办。抛弃超过的请求，显示当前访问数过大，请稍后再试。这种比显示服务器崩溃了友好的多。
再例如秒杀，由于队列的先进先出特性，第一个请求，那就是第一个被拿到，所以就抢到了，后面的也只能告诉你，秒杀失败，手速不够啊。
二、概述
重要概念：
消息代理和目的地
当消息发送者发送消息后，由消息代理接管，消息代理保证消息传递到指定目的地。
消息队列两种主要形式的目的地：
1.队列：点对点消息通信
2.主题：发布/订阅 消息通信
点对点式：
消息发送者发送消息，消息代理将其放入一个队列中，消息接收者从队列中获取消息内容，消息读取后被移除队列。
注意：消息只有唯一的发送者和接收者，但并不是说只能有一个接收者，多个接收者获取，谁先拿到就是谁的，拿到后队列中的消息消失，其他接受者就拿不到了。
发布订阅式：
发送者发送消息到主题，多个接收者监听订阅这个主题，那么就会在消息到达的同时接收到消息。
JMS Java消息服务
sun公司的，基于JVM消息代理的规范。ActiveMQ、HornetMQ是JMS实现
AMQP
高级消息队列协议，也是消息代理的规范，兼容JMS
RabbitMQ是AMQP的实现
两种规范的区别如下：
三、RabbitMQ概念
RabbitMQ简介：
RabbitMQ是一个由erlang开发的AMQP的开源实现
核心概念：
Message
消息，消息是不具名的，它由消息头和消息体组成，消息体是不透明的，而消息头则是由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等
Publisher
消息的生产者，也是一个向交换器发布消息的客户端的应用程序。
Exchange
交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。Exchange有4中类型：direct（默认）、fanout、topic和headers，不同类型的Exchange转发消息的策略有所区别
Queue
消息队列，用来保存消息知道发送给消费者，它是消息的容器，也是消息的终点，一个消息可投入一个或者多个队列，消息一直在队列里面，等待消费者连接到这个队列将其取走。
Binding
绑定，用于消息队列和交换器之间的关联，一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。
Exchange和Queue的绑定可以是多对多的关系
Connection
网络链接，比如一个TCP连接
Channel
信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP链接内的虚拟连接，AMQP命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成，因为对于操作系统来说建立和销毁TCP都是非常昂贵的开销，所以引入了信道的概念，以复用一条TCP的连接
工作流程：
四、Dockeran安装RabbitMQ
docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672
rabbitmq:management
4369,25672(Erlang发现&集群端口)
5672,5671(AMQP端口)
15672 (web管理后台端口)
61613,61614(STOMP协议端口)
1883,8883(MQTT协议端口)
安装成功后 访问ip加端口
例如我的：
http://192.168.137.4:15672
账号：guest
密码：guest
界面详情：
五、交换机类型
Direct: 精确匹配，完全匹配设置的路由
fanout:给绑定的队列都发消息
topic：匹配路由，由 . 分割，例如路由键为usa.news则usa.#和#.news都可以拿到。#代表一个或多个单词，*代表一个单词
创建一个交换机：
创建一个队列，然后点击交换机的名字，进入详情进行绑定
绑定交换机与队列，并指定之间的路由键,可根据交换机类型，设置路由匹配规则
发送消息：
点击队列名字，进入详情
六、集成SpringBoot
导入依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>2.1.6.RELEASE</version>
</dependency>
开启功能
@EnableRabbit
配置文件：
spring:
cloud:
rabbitmq:
host: ${ipAddr}
port: 5672
# 指定虚拟主机
virtual-host: /
username: guest
password: guest
# 开启发送端确认
publisher-confirms: true
# 开启发送端消息抵达队列确认
publisher-returns: true
# 只要抵达队列，以异步优先回调我们这个 returnconfirm
template:
mandatory: true
# 手动ack消息 消息不容易丢失
listener:
simple:
acknowledge-mode: manual
测试：
创建交换机：
/**
* 1、创建Exchange Queue Binding
* 2、发送消息
*/
@Test
public void createExchange() {
DirectExchange exchange = new DirectExchange("hello-java-exchange", true, false);
amqpAdmin.declareExchange(exchange);
log.info("\nExchange [" + exchange.getName() + "] 创建成功");
}
创建队列
@Test
public void createQueue() {
// 持久化：true 是否排他：false 是否自动删除：false
Queue queue = new Queue("hello-java-queue", true, false, false);
amqpAdmin.declareQueue(queue);
log.info("\nQueue [" + queue.getName() + "] 创建成功");
}
绑定：
/**
* 目的地 目的地类型 交换机 路由键
* String destination, DestinationType destinationType, String exchange, String routingKey,
* @Nullable Map<String, Object> arguments
*/
@Test
public void bindIng() {
Binding binding = new Binding("hello-java-queue", Binding.DestinationType.QUEUE, "hello-java-exchange", "hello.java", null);
amqpAdmin.declareBinding(binding);
log.info("\n[" + binding.getExchange() + "] 与 [" + binding.getDestination() + "] 绑定成功");
}
发送消息：
/**
* 发送的消息是一个对象 必须实现序列化
*/
@Test
public void sendMessageTest(){
String msg="hello word";
rabbitTemplate.convertAndSend("hello-java-exchange","hello.java",msg);
}
发送消息为对象时：
加入配置文件：
package com.qixi.magpie.order.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
@Slf4j
@Configuration
public class MyRabbitConfig {
// @Autowired
private RabbitTemplate rabbitTemplate;
@Primary
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
this.rabbitTemplate = rabbitTemplate;
rabbitTemplate.setMessageConverter(messageConverter());
initRabbitTemplate();
return rabbitTemplate;
}
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
/**
* 1.设置确认回调： ConfirmCallback
* 先在配置文件中开启 publisher-confirms: true
* @PostConstruct: MyRabbitConfig对象创建完成以后 执行这个方法
*
* 2.消息抵达队列的确认回调
* 开启发送端消息抵达队列确认
* publisher-returns: true
* 只要抵达队列，以异步优先回调我们这个 returnconfirm
* template:
* mandatory: true
* 3.消费端确认(保证每一个消息被正确消费才可以broker删除消息)
* 1.默认是自动确认的 只要消息接收到 服务端就会移除这个消息
*
* 如何签收:
* 签收: channel.basicAck(deliveryTag, false);
* 拒签: channel.basicNack(deliveryTag, false,true);
* 配置文件中一定要加上这个配置
* listener:
* simple:
* acknowledge-mode: manual
*/
// @PostConstruct
public void initRabbitTemplate(){
/**
* 设置确认回调
* correlationData: 消息的唯一id
* ack： 消息是否成功收到
* cause：失败的原因
*/
rabbitTemplate.setConfirmCallback((correlationData, ack , cause) -> log.info("\n收到消息: " + correlationData + "\tack: " + ack + "\tcause： " + cause));
/**
* 设置消息抵达队列回调：可以很明确的知道那些消息失败了
*
* message: 投递失败的消息详细信息
* replyCode: 回复的状态码
* replyText: 回复的文本内容
* exchange: 当时这个发送给那个交换机
* routerKey: 当时这个消息用那个路由键
*/
rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routerKey) -> log.error("Fail Message [" + message + "]" + "\treplyCode: " + replyCode + "\treplyText:" + replyText + "\texchange:" + exchange + "\trouterKey:" + routerKey));
}
}
messageConverter()用于转为json格式：
/**
* 发送的消息是一个对象 必须实现序列化
*/
@Test
public void sendMessageTest(){
OrderEntity entity = new OrderEntity();
entity.setId(1L);
entity.setCommentTime(new Date());
entity.setCreateTime(new Date());
entity.setConfirmStatus(0);
entity.setReceiverCity("重庆");
}
接收消息：
springboot接收消息
使用@RabbitListener注解必须要有@EnableRabbit
/**
* @Author li
* @Description queues 申明监听的队列，可以是多个
* @Date 2021/7/9/009
* @Param []
* @return void
**/
@RabbitListener(queues = {"hello-java-queue"})
public void listener(Object messgae){
System.out.println("收到消息："+messgae+"===>类型是："+messgae.getClass());
}
然后再次点击发送消息：
控制台打印结果：
收到消息：(Body:'{"id":2,"username":null,"orderNum":null,"totalAmount":null,"payAmount":null,"freightAmount":null,"sourceType":null,"payType":null,"status":null,"deliveryCompany":null,"deliverySn":null,"autoConfirmDay":null,"intimacy":null,"billType":null,"billHeader":null,"billContent":null,"billReceiverPhone":null,"billReceiverEmail":null,"receiverName":null,"receiverPhone":null,"receiverPostCode":null,"receiverProvince":null,"receiverRegion":null,"receiverCity":"重庆","receiverDetailAddress":null,"note":null,"confirmStatus":0,"deleteStatus":null,"paymentTime":null,"deliveryTime":null,"receiveTime":null,"commentTime":1625823135893,"modifyTime":null,"giftStatus":null,"give":null,"createTime":1625823135893,"updateTime":null}' MessageProperties [headers={__TypeId__=com.qixi.magpie.order.entity.OrderEntity}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=hello-java-exchange, receivedRoutingKey=hello.java, deliveryTag=2, consumerTag=amq.ctag-Gje7WEjhHGYm5rx3jIvFhg, consumerQueue=hello-java-queue])===>类型是：class org.springframework.amqp.core.Message
注意：
/**
* 1.Message message: 原生消息类型 详细信息
* 2.T<发送消息的类型> OrderEntity orderEntity [Spring自动帮我们转换]
* 3.Channel channel: 当前传输数据的通道
*
* // 同一个消息只能被一个人收到
* // 多个消息，只能等当前的消息处理接收完，才能接收下一个
*
* @RabbitListener： 只能标注在类、方法上 配合 @RabbitHandler
* @RabbitHandler: 只能标注在方法上 [重载区分不同的消息]
* //应用接受同个队列，但是发送不同类型的消息时，两种注解配合使用
*/
例如：
@RabbitListener(queues = {"hello-java-queue"})
@Service("detailsService")
public class DetailsServiceImpl extends ServiceImpl<DetailsDao, DetailsEntity> implements DetailsService {
@Override
public PageUtils queryPage(Map<String, Object> params) {
IPage<DetailsEntity> page = this.page(
new Query<DetailsEntity>().getPage(params),
new QueryWrapper<DetailsEntity>()
);
return new PageUtils(page);
}
/**
* 1.Message message: 原生消息类型 详细信息
* 2.T<发送消息的类型> OrderEntity orderEntity [Spring自动帮我们转换]
* 3.Channel channel: 当前传输数据的通道
*
* // 同一个消息只能被一个人收到
*
*
* @RabbitListener： 只能标注在类、方法上 配合 @RabbitHandler
* @RabbitHandler: 只能标注在方法上 [重载区分不同的消息]
*/
@RabbitHandler
public void receiveMessageA(Message message, OrderEntity orderEntity, Channel channel){
System.out.println("接受到消息: " + message + "\n内容：" + orderEntity);
try {
Thread.sleep(200);
} catch (InterruptedException e) { }
// 这个是一个数字 通道内自增
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
// 只签收当前货物 不批量签收
channel.basicAck(deliveryTag, false);
// deliveryTag: 货物的标签 multiple: 是否批量拒收 requeue: 是否重新入队
// channel.basicNack(deliveryTag, false,true);
// 批量拒绝
// channel.basicReject();
} catch (IOException e) {
System.out.println("网络中断");
}
System.out.println(orderEntity.getReceiverName() + " 消息处理完成");
}
@RabbitHandler
public void receiveMessageB(Message message, PaymentInfoEntity paymentInfoEntity, Channel channel){
System.out.println("接受到消息: " + message + "\n内容：" + paymentInfoEntity);
try {
Thread.sleep(2000);
} catch (InterruptedException e) { }
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
channel.basicAck(deliveryTag, false);
} catch (IOException e) {
System.out.println("网络中断");
}
System.out.println(paymentInfoEntity.getId() + " 消息处理完成");
}
}
@RabbitListener标注在类上，指定需要监听的队列
@RabbitHandler标注在方法上，区分类型。
七、RabbitMQ消息确认机制-可靠抵达
确保消息不丢失，可靠抵达，可以使用事务消息，性能下降250倍，为此引入确认机制
publisher confirmCallback确认模式
publisher returnCallback 未投递到queue 退回模式
consumer ack机制
ConfirmCallback
spring.rabbitmq.publisher-confirms=true
? -在创建connectionFactory的时候设置PublisherConfirms(true)选项，开启confirmCallback。
? -CorrelationData:用来表示当前消息唯一性。
? -消息只要被borker接收到就会执行confirmCallback。如果是cluster模式，需要所有broker接收到才会调用confirmCallback.
? -被broker接收到只能表示message已经到达服务器，并不能保证消息一定会被投递到目标queue里，所以需要用到接下来的returnCallback
例如：
spring:
rabbitmq:
host: 192.168.137.4
port: 5672
# 指定虚拟主机
virtual-host: /
username: guest
password: guest
# 开启发送端确认
publisher-confirms: true
# 开启发送端消息抵达队列确认
publisher-returns: true
# 只要抵达队列，以异步优先回调我们这个 returnconfirm
template:
mandatory: true
# 手动ack消息 消息不容易丢失
listener:
simple:
acknowledge-mode: manual
还需要配置文件设置确认回调：
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
@Slf4j
@Configuration
public class MyRabbitConfig {
// @Autowired
private RabbitTemplate rabbitTemplate;
@Primary
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
this.rabbitTemplate = rabbitTemplate;
rabbitTemplate.setMessageConverter(messageConverter());
initRabbitTemplate();
return rabbitTemplate;
}
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
/**
* 1.设置确认回调： ConfirmCallback
* 先在配置文件中开启 publisher-confirms: true
* @PostConstruct: MyRabbitConfig对象创建完成以后 执行这个方法
*
* 2.消息抵达队列的确认回调
* 开启发送端消息抵达队列确认
* publisher-returns: true
* 只要抵达队列，以异步优先回调我们这个 returnconfirm
* template:
* mandatory: true
* 3.消费端确认(保证每一个消息被正确消费才可以broker删除消息)
* 1.默认是自动确认的 只要消息接收到 服务端就会移除这个消息
*
* 如何签收:
* 签收: channel.basicAck(deliveryTag, false);
* 拒签: channel.basicNack(deliveryTag, false,true);
* 配置文件中一定要加上这个配置
* listener:
* simple:
* acknowledge-mode: manual
*/
// @PostConstruct
public void initRabbitTemplate(){
/**
* 设置确认回调
* correlationData: 消息的唯一id
* ack： 消息是否成功收到
* cause：失败的原因
*/
rabbitTemplate.setConfirmCallback((correlationData, ack , cause) -> log.info("\n收到消息: " + correlationData + "\tack: " + ack + "\tcause： " + cause));
/**
* 设置消息抵达队列回调：可以很明确的知道那些消息失败了
*
* message: 投递失败的消息详细信息
* replyCode: 回复的状态码
* replyText: 回复的文本内容
* exchange: 当时这个发送给那个交换机
* routerKey: 当时这个消息用那个路由键
*/
rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routerKey) -> log.error("Fail Message [" + message + "]" + "\treplyCode: " + replyCode + "\treplyText:" + replyText + "\texchange:" + exchange + "\trouterKey:" + routerKey));
}
}
收到效果：
ReturnCallback
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true
? -confrim模式只能保证消息到达broker，不能保证消息准确投递到目标queue里。在有些业务场景下，我们需要保证消息一定要投递到目标queue里，此时就需要用到return退回模式。
? -这样如果未能投递到目标queue里将调用returnCallback,可以记录下详细到投递数据，定期的巡检或者西东纠错都需要这些数据
配置文件如上方配置的yml和config。
Ack
消费者获取到消息，成功处理，可以回复Ack给Broker
? -basic.ack用于肯定确认；broker将移除此消息
? -basic.nack用于否定确认；可以指定broker是否丢弃此消息，可以批量
? -basic.reject用于否定确认，同上，但不能批量
默认，消息被消费者收到，就会从broker的queue中移除
queue无消费者，消息依然会被存储，直到消费者消费
消费者收到消息，默认自动ack。但是如果无法确定此消息是否被处理完成或者成功处理。我们可以开启手动ack模式
? -消息处理成功，ack()，接受下一个消息，此消息broker就会移除
? -消息处理失败，nack()/rejeck()，重新发送给其他人进行处理，或者容错处理后ack
? -消息一直没有调用ack/nack方法，broker认为消息正在被处理，不会投递给别人，此时客户端断开，消息不会被broker移除，会投递给别人。
开启手动确认：
spring.rabbitmq.listener.simple.acknowledge-mode= manual
如上方yml配置文件
签收(与拒签)：
签收： channel.basicAck(deliveryTag, false);
拒绝： channel.basicNack(deliveryTag, false,true);
// 这个是一个数字 通道内自增
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
// 只签收当前货物 不批量签收
channel.basicAck(deliveryTag, false);
// deliveryTag: 货物的标签 multiple: 是否批量拒收 requeue: 是否重新入队
// channel.basicNack(deliveryTag, false,true);
// 批量拒绝
// channel.basicReject();
} catch (IOException e) {
System.out.println("网络中断");
}