RabbitMQ
1. 安装教程
- windows平台
Erlang安装:Erlang安装_erlang安装教程-CSDN博客
Erlang和RabbitMQ版本兼容:windows安装rabbitmq和环境erlang(最详细版,包括对应关系,安装错误解决方法)_erlang和rabbitmq关系-CSDN博客
- linux
docker平台安装rabbitmq:
Linux平台使用Docker安装RabbitMQ - ooyhao - 博客园 (cnblogs.com)
Docker中安装RabbitMQ - niceyoo - 博客园 (cnblogs.com)
Docker下RabbitMQ安装 - work hard work smart - 博客园 (cnblogs.com)
docker run -d --hostname my-rabbitmq \
-p 15672:15672 -p 5672:5672 \
--name my-rabbitmq \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=admin \
rabbitmq:3.8.9-management
关于参数:
- -d:以守护进程方式在后台运行
- --hostname:设置容器的主机名称,仅本机可见。这种方式是写到 /etc/hostname ,以及 /etc/hosts 文件中,作为容器主机IP的别名,并且将显示在容器的bash中。不过这种方式设置的主机名将不容易被容器之外可见。这将不会出现在 docker ps 或者 其他的容器的 /etc/hosts 文件中。
- --name:指定容器名
- -p:指定服务运行的端口。
- RABBITMQ_DEFAULT_USER:用于设置登陆控制台的用户名,这里我设置 admin
- RABBITMQ_DEFAULT_PASS:用于设置登陆控制台的密码,这里我设置 admin容器启动成功后,可以在浏览器输入地址:http://ip:15672/访问控制台
额外说一下 rabbitmq 这几个端口的作用:
- 4369:EPMD( Erlang Port Mapper Daemon)端口号,在 Erlang 集群中相当于 dns 的作用
- 5672:client 端通信端口
- 15672:web管理界面端口
2. 详细笔记
- 黑马详细笔记:入门
黑马程序员-RabbitMQ | WeiBlog (gitee.io)
- 其他详细笔记
新RabbitMQ精讲 提升工程实践能力 培养架构思维 (yuque.com)
手把手教你SpringBoot集成消息服务中间件RabbitMQ - 你樊不樊 - 博客园 (cnblogs.com)
3. rabbitmq 6种队列模式
RabbitMQ - 随笔分类 - niceyoo - 博客园 (cnblogs.com)
黑马程序员-RabbitMQ | WeiBlog (gitee.io)
rabbitmq详细实例 - JiuYou2020 - 博客园 (cnblogs.com)
4. Rabbitmq模块实现
来源:个人做了一个Spring Boot整合RabbitMQ的笔记
含工作、发布订阅、路由、主题模式示例
RabbitMQ消息可靠投递,可靠消费。完全可以在工作中直接cv使用。
github地址:https://github.com/WENZIZZHENG/spring-boot-demo
gitee地址:https://gitee.com/BiMon/spring-boot-demo
5. 学习遇到问题
使用Idea创建一个父子SpringBoot项目 - 飘杨...... - 博客园 (cnblogs.com)
6. springboot 集成rabbitmq
1. 生产者
- 导入坐标
详解Spring Boot 项目中的 parent_java_脚本之家 (jb51.net)
SpringBoot到底是什么?如何理解parent、starter、引导类以及内嵌Tomcat? - 知乎 (zhihu.com)
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.6.13</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
- rabbitmq配置
spring:
rabbitmq:
host: 8.*.*.213
username: admin
password: admin
virtual-host: /rabbitmq_springboot
port: 5672
- rabbitmqConfig编写
@Configuration
public class RabbitConfig {
public static final String EXCHANGE_NAME="boot_topic_exchange";
public static final String QUEUE_NAME="boot_queue";
//交换机
@Bean("bootExchange")
public Exchange bootExchange(){
return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
}
//2.Queue队列
@Bean("bootQueue")
public Queue bootQueue(){
return QueueBuilder.durable(QUEUE_NAME).build();
}
//3.队列和交换就绑定关系 Binding
/*
1.知道哪个队列
2.知道哪个交换机
3.routing key
*/
@Bean
public Binding bindingQueueExchange(@Qualifier("bootQueue")Queue queue,@Qualifier("bootExchange")Exchange exchange){
return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
}
}
- 测试
@SpringBootTest
class BootproducerApplicationTests {
@Resource
private RabbitTemplate rabbitTemplate;//注入rabbitmq模板类
@Test
void springbootRabbitMq(){
rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME,"boot.hahah","boot mq hello....");
}
}
2. 消费者
- 导入依赖
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.6.13</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
spring:
rabbitmq:
host: 8.*.*.213
username: admin
password: admin
virtual-host: /rabbitmq_springboot
port: 5672
- 编写消费者
@Component
public class RabbitMQListener {
@RabbitListener(queues = "boot_queue")
public void ListenerQueue(Message message) {
System.out.println(message);
System.out.println(message.getBody());
}
}
7.rabbitmq 高级篇
1. 生产者消息可靠投递:confirm机制
- confirm 确认流程
RabbitMQ消息可靠性传输 - 一步一年 - 博客园 (cnblogs.com)
- 代码编写
1. 生产者配置yml 文件
spring:
rabbitmq:
publisher-returns: true # return 确认机制
template:
mandatory: true
publisher-confirm-type: simple
publisher-returns: 开启消息从exhcange路由到queue的回调,只有路由失败时才会触发回调
mandatory: 为true时,如果exchange根据routingKey将消息路由到queue时找不到匹配的queue,触发return回调,为false时,exchange直接丢弃消息。
publisher-confirm-type: 表示确认消息的类型,分别有none、correlated、simple这三种类型。
-
none:表示禁用发布确认模式
-
correlated:表示消息成功到达Broker后触发ConfirmCalllBack回调
-
simple: simple模式下如果消息成功到达Broker后一样会触发ConfirmCalllBack回调,发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果
RabbitMQ相关--消息确定机制_publisher-confirm-type-CSDN博客
2. 编写配置类
/* description
* 1.RabbitTemplate.ConfirmCallback 消息被正常发送到交换机, 则会调用该方法, 自动回调
* 2.RabbitTemplate.ReturnsCallback 消息被交换机正常发送到队列,则会回调改方法, 自动回调
*/
@Component
@Slf4j
public class RabbitMQConfirmAndReturnConfig implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void initMethod(){
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnsCallback(this);
}
/**
* @description:
*
* @param correlationData: 相关数据,对象中有一个id的属性,用来表示消息的唯一标识
* @param ack: true:表示正常发送成功
* @param cause 表示消息发送失败原因
* @author life
* @date 2024-01-19 16:55
* @return void
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if(ack){
System.out.println("生产者到交换机成功");
}else{
System.out.println("消息发送到交换机失败");
}
}
//RabbitTemplate.ReturnsCallback
/**
* @description:
*消息发功成功,则会不调用改方法, 消息失败 则调用该方法
* @param returnedMessage:
* -message: 发送消息内容
* - replyCode: 回应码
* - replyText: 回应内容
* - exchange:交换机
* - routingKey: 路由键
* @author life
* @date 2024-01-19 16:58
* @return void
*/
@Override
public void returnedMessage(ReturnedMessage returnedMessage) {
System.out.println("交换机到队列失败");
}
}
参考博客
视频:1.RabbitMQ消息确认机制介绍_哔哩哔哩_bilibili
RabbitMQ消息可靠性传输 - 一步一年 - 博客园 (cnblogs.com)
RabbitMQ相关--消息确定机制_publisher-confirm-type-CSDN博客
2. 消费者如何保证消息不丢失:手动提交ack
- 存在情况
消费者接收到消息,但是还未处理或者还未处理完,此时消费者进程挂掉了,比如重启或者异常断电等,此时mq认为消费者已经完成消息消费,就会从队列中删除消息,从而导致消息丢失。
- 解决办法
RabbitMQ默认是自动ack的,此时需要将其修改为手动ack,也即自己的程序确定消息已经处理完成后,手动提交ack,此时如果再遇到消息未处理进程就挂掉的情况,由于没有提交ack,RabbitMQ就不会删除这条消息,而是会把这条消息发送给其他消费者处理,但是消息是不会丢的。
- 具体实现办法
消费者application.yaml:配置信息
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: manual # 手动ack
prefetch: 5 # 消峰 最多有5个
参数
消费者手动确认
@Component
public class RabbitMQListener {
@RabbitListener(queues = "boot_queue")
public void ListenerQueue(Message message, Channel channel) throws IOException {
try{
System.out.println(message);
int a=1/0;
// 确认回复收到 true:批量确认消息 false 单独确认消息
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}catch (Exception e){
// 参数2 true 批量确认消息 false 单独确认消息
// 参数3 true 重发 插入末尾 false 队列丢弃
channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
}
}
}
对于channel.basicNack方法的第三个参数,表示消息nack后是否返回队列,如果设置为true,表示返回队列,此时消息处于队列头部,消费者会一直处理该消息,影响后续消息的消费,设置为false时表示不返回队列,此时如果设置有DLX(死信队列),那么消息会进入DLX中,后续再对该消息进行相应的处理,如果没有设置DLX,此时消息就会被丢弃。
3. 消费端限流
- 演示
- 实现
applicaton.yaml配置
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: manual # 手动确认
prefetch: 5 # 消减峰值
- 注意事项
4. 死信队列
- 死信队列
死信,Dead Letter,一种消息机制,当消费者去消费队列中的消息时,如果队列中的消息出现了以下的情况:
- 消费端执行nack或者reject时,设置requeue=false;: 消费者拒绝接受
- 消息在队列中的时间超过设置的TTL(Time To Live)时间;:消息过期
- 队列中消息的数量超过设置的最大数量;: 队列满,再添加消息
那么这些消息就可以被称之为死信消息,在配置了死信队列的情况下,死信消息会进入死信队列,如果没有配置死信队列,这些死信消息会被丢弃。
死信队列和普通队列
死信交换机可以是fanout、direct、topic等类型,和普通交换机并无不同;
死信交换机要绑定要业务队列上才会生效;
给死信交换机绑定的队列称之为死信队列,其实就是普通的队列,没有任何特殊之处
- 死信队列实现
架构图
生产者
@Configuration
public class RabbitDLConfig {
public static final String EXCHANGE_NAME="boot_topic_exchange";//普通交换机
public static final String DL_EXCHANGE_NAME="topic_exchange_deadLetter";//死信交换机
public static final String QUEUE_NAME="boot_queue";//普通队列
public static final String DL_QUEUE_NAME="queue_deadLetter";//死信交换机
//普通队列
//1. 交换机
@Bean("bootExchange")
public Exchange bootExchange(){
return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
}
//2.Queue队列
@Bean("bootQueue")
public Queue bootQueue(){
return QueueBuilder
.durable(QUEUE_NAME)
.deadLetterExchange(DL_EXCHANGE_NAME)//绑定死信交换机
.deadLetterRoutingKey("dl.orginal")// 指定路由键
.build();
}
//3.队列和交换就绑定关系 Binding
/*
1.知道哪个队列
2.知道哪个交换机
3.routing key
*/
@Bean
public Binding bindingQueueExchange(@Qualifier("bootQueue")Queue queue, @Qualifier("bootExchange")Exchange exchange){
return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
}
// 死信队列
//1. 死信交换机
@Bean("deadLetterExchange")
public Exchange deadLetterExchange(){
return ExchangeBuilder.topicExchange(DL_EXCHANGE_NAME).durable(true).build();
}
//2.死信Queue队列
@Bean("deadLetterQueue")
public Queue deadLetterQueue(){
return QueueBuilder.durable(DL_QUEUE_NAME).build();
}
//3.队列和交换就绑定关系 Binding
/*
1.知道哪个队列
2.知道哪个交换机
3.routing key
*/
//死信队列绑定死信交换机
@Bean
public Binding bindingDLQueueExchange(@Qualifier("deadLetterQueue")Queue queue, @Qualifier("deadLetterExchange")Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("dl.#").noargs();
}
}
普通队列消费者
@Component
public class RabbitMQListener {
@RabbitListener(queues = "boot_queue")
public void ListenerQueue(Message message, Channel channel) throws IOException {
System.out.println("我是普通消费者:"+message.getBody());
}
}
死信队列消费者
// 死信队列
@Component
public class DeadLetterRQListener {
@RabbitListener(queues={"queue_deadLetter"})// 绑定死信队列
public void ListenerDeadQueue(Message message, Channel channel){
System.out.println("我是死信队列消费者:"+message.getBody());
}
}
生产者测试 消息过期:
@SpringBootTest
public class DeadLetterQueueTest {
@Resource
private RabbitTemplate rabbitTemplate;
@Test
public void testDeadQueue(){
MessagePostProcessor messagePostProcessor = message -> {
//1.设置message的信息
message.getMessageProperties().setExpiration("1");//消息的过期时间
//2.返回该消息
return message;
};
for(int i=0;i<10;i++){
String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd"));
String msg="生产者发送第"+i+"条消息给消费者"+"时间:"+time;
if(i%2==0){ rabbitTemplate.convertAndSend(RabbitDLConfig.EXCHANGE_NAME,"boot.hahah",msg);
}else{ rabbitTemplate.convertAndSend(RabbitDLConfig.EXCHANGE_NAME,"boot.hahah",msg, messagePostProcessor);
}
}
}
}
私信队列总结
死信队列更多的是用来保证消息的可靠性,主要用于比较重要的队列,用以确保未被正确消费的消息不会丢失,其实也可以不用死信队列,在消费端出现异常时,可以将消息从当前队列ack掉,再将其发送到其他队列,然后再单独处理其他队列
RabbitMQ死信队列 - 一步一年 - 博客园 (cnblogs.com)
5.延迟队列
- 什么是延迟队列
延时队列中的元素是希望在指定时间到了以后或之前取出和处理,简单来说,延时队列就是用来存放需要在指定时间被处理的元素的队列。
- 使用场景
- 订单在十分钟之内未支付则自动取消
- 新创建的店铺,如果在十天内都没有上传过商品,则自动发送消息提醒。
- 用户注册成功后,如果三天内没有登陆则进行短信提醒。
- 用户发起退款,如果三天内没有得到处理则通知相关运营人员。
- 预定会议后,需要在预定的时间点前十分钟通知各个与会人员参加会议
如何将未付款的订单取消?
方法 一 :定时任务轮询查找数据。数据量小的时候可以
方法二: 消息队列延时队列解决。
- 延时队列实现一:死信队列+ttl 过期时间
死信队列+ttl 过期时间
- 延时队列实现二:延时队列插件
1. 下载插件
docker安装RabbitMQ及安装延迟插件的详细过程_docker_脚本之家 (jb51.net)
#刚刚上传的插件拷贝到容器内plugins目录下
# docker cp 当前插件地址 容器名字:/plugins
docker cp /root/rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez my-rabbitmq:/plugins
# 进入容器
docker exec -it my-rabbitmq /bin/bash
2. 启用插件,重启容器
# 启动延迟插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
[root@iZ2vc2rht08hck1x6us154Z ~]# docker restart my-rabbitmq
3. 查看是否安装成功
4.交换机配置修改
5. 生产者 setExpiration 改成setDeLay
博客链接
八、RabbitMq死信队列与延迟队列 - hp柠檬茶 - 博客园 (cnblogs.com)
MQ高级-11.延迟消息-死信交换机_哔哩哔哩_bilibili
6.幂等性:解决重复消费
- 什么是幂等性
用户对于同一操作发起的一次请求或者多次请求的结果是一致的。
例子:退款业务出现问题,多次退款 恢复多次余额
- 消息重复消费
消费者在消费 MQ 中的消息时,MQ 已把消息发送给消费者,消费者在给 MQ 返回 ack 时网络中断,故 MQ 未收到确认信息,该条消息会重新发给其他的消费者,或者在网络重连后再次发送给该消费者,但实际上该消费者已成功消费了该条消息,造成消费者消费了重复的消息。
- 幂等性解决办法
1.唯一消息id
- MQ 消费者的幂等性的解决一般使用全局 ID 或者写个唯一标识比如时间戳 或者 UUID 或者订单消费者消费 MQ 中的消息也可利用 MQ 的该 id 来判断,
- 或者可按自己的规则生成一个全局唯一 id,每次消费消息时用该 id 先判断该消息是否已消费过
- 消费端的幂等性保障
a. 唯一 ID+指纹码机制,利用数据库主键去重。
b. 利用 redis 的原子性去实现:利用 redis 执行 setnx 命令,天然具有幂等性。从而实现不重复消费
博客链接
十、RabbitMQ 幂等性 - hp柠檬茶 - 博客园 (cnblogs.com)
7.docker搭建rabbitmq 集群
- 拉取rabbitmq镜像
docker pull rabbitmq:3.8.9-management
- 启动多个rabbitmq容器
主机1
docker run -d --hostname my-rabbitmq \
-p 15672:15672 -p 5672:5672 \
--name my-rabbitmq \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=admin \
-e RABBITMQ_ERLANG_COOKIE='admin'\
rabbitmq:3.8.9-management
- -e RABBITMQ_ERLANG_COOKIE='rabbit_cluster' 设置rabbitmq的cookie,该值可以任意设置,只需要三个容器保持一致即可 :集群之间相互通信 cookie一致
- --link 容器名:主机名 作用不同主机之间相互通信
主机2
docker run -d --hostname my-rabbitmq1 \
-p 15673:15672 -p 5673:5672 \
--name my-rabbitmq1 \
--link my-rabbitmq:my-rabbitmq \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=admin \
-e RABBITMQ_ERLANG_COOKIE='admin'\
rabbitmq:3.8.9-management
主机2和主机1 连接
主机3
docker run -d --hostname my-rabbitmq2 \
-p 15674:15672 -p 5674:5672 \
--name my-rabbitmq2 \
--link my-rabbitmq:my-rabbitmq \
--link my-rabbitmq1:my-rabbitmq1 \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=admin \
-e RABBITMQ_ERLANG_COOKIE='admin'\
rabbitmq:3.8.9-management
主机3和主机2 和主机1链接
- 配置集群
重置节点一
docker exec -it my-rabbitmq /bin/bash
rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl start_app
exit
节点一名称:rabbit@my-rabbitmq
重置节点二
docker exec -it my-rabbitmq1 /bin/bash
rabbitmqctl stop_app && rabbitmqctl reset
rabbitmqctl join_cluster rabbit@my-rabbitmq
rabbitmqctl start_app
exit
节点一名称:rabbit@my-rabbitmq1
重置节点三
docker exec -it my-rabbitmq2 /bin/bash
rabbitmqctl stop_app && rabbitmqctl reset
rabbitmqctl join_cluster rabbit@my-rabbitmq1
rabbitmqctl start_app
exit
- 查看集群状态
rabbitmqctl cluster_status
- 浏览器查看集群状态
- 关闭其他节点
多台服务器配置rabbit集群: Docker环境RabbitMq配置SSL - 一步一年 - 博客园 (cnblogs.com)
8. rabbitmq
- 封装消息
/* 发送消息 */
//1. 设置消息的编码和内容格式
MessageProperties properties = new MessageProperties();
properties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
properties.setContentEncoding("utf-8");
//2. 消息对象转成json字符串
String jsonMsg = JSON.toJSONString(messageDTO);
//3. 将所需要发送的消息 ,以及消息的格式封装成Message对象
Message sendMsg = new Message(jsonMsg.getBytes(StandardCharsets.UTF_8),properties);
//4. Correlateion 设置returnMessage失败返回内容
CorrelationData correlationData = new CorrelationData();
ReturnedMessage returnedMessage = new ReturnedMessage(sendMsg,300,"餐厅收到订单消息,发送给订单服务队列",
RabbitConfig.RESTAURANT_EXCHANGE,
"key.order");
correlationData.setReturned(returnedMessage);
correlationData.setId(messageDTO.getOrderID()); //设置为订单ID
//5. 生产者发送消息
rabbitTemplate.convertAndSend(RabbitConfig.RESTAURANT_EXCHANGE,"key.order",sendMsg,correlationData);
// 6. 消费者手动确认消息答
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false); //手动消息应答
面试题
rabbitMQ 消息顺序性、消息幂等性、消息不丢失、最终一致性、补偿机制、消息队列设计_消息补偿机制-CSDN博客
相关参考
视频参考
rabbitmq 视频:21rabbit集群搭建_哔哩哔哩_bilibili
面试题:RabbitMQ面试题详解 | 一口气看完16个面试必问的rabbitmq面试核心知识点,让你面试少走99%的弯路!_哔哩哔哩_bilibili
博客推荐
RabbitMQ消息可靠性传输 - 一步一年 - 博客园 (cnblogs.com)
标签:队列,rabbitmq,死信,RabbitMQ,public,消息 From: https://www.cnblogs.com/life1314/p/17988136