devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

消息队列：RabbitMQ_rabbitmq ack 是单机的处理速度么

作者：从前慢现在也慢 | 2024-08-06 19:48:34

赞

踩

rabbitmq ack 是单机的处理速度么

1 相关概念

1.1什么是MQ(message queue)

本质是个队列，FIFO先入先出，存放的内容是message，而且还是一种跨进程的通信机制，用于传递上下游消息，在互联网架构中MQ是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务，使用MQ之后，消息发送上游只需要依赖MQ不需要依赖其他服务。

1.2.为什么用MQ

流量削峰

如果系统最多可以处理10000并发量，平时使用肯定时绰绰有余，1秒就可以反馈结果，但是如果是高峰期可能同时出现20000或者更多的并发量，超过10000的我们只能禁止他进行该操作。使用MQ做缓冲之后我们可以把每秒的操作分配到一段时间内来处理，这样会造成操作之后几十秒以后才会收到结果反馈，但是比限制操作要有好很多。也可以防止后端系统宕机。但是相应的访问速度会下降。

应用解耦

一个应用中有多个系统，如果A系统宕机，那么与其耦合调用的B系统也会发生异常，最终造成整个应用的宕机，而使用了MQ之后，如果A系统鼓掌，B系统的请求被缓存在消息队列种，当A系统恢复后，再从消息队列种拿到请求进行处理，这样可以避免很多因系统调用造成的问题。

异步处理

当A系统调用了B系统处理一个很长时间才可以完成的请求，并且A需要知道B是否处理完成，以前一般有两种方式：第一种是A过一段时间去调用一下B提供的查询API来确认，第二种是B提供一个calllback的api当B处理完成之后调用一下A的callback，这两种方式不是很优雅。但是使用消息总线可以解决这个问题，当A调用B后，去监听B的处理完成的消息，B处理完成后，会向MQ发送一个消息，由MQ转发给A，当A监听到后就代表B已经处理完成，这样既避免了A重复调用B的查询接口，也不用提供callback的API，同样B也不用做这些操作，A还可以及时的到B的处理反馈。

1.3.MQ的分类

ActiveMQ

比较老的MQ

优点：

单机吞吐量可达万级，时效性是ms级，可用性高，基于主从框架实现高可用性，消息可靠性，较低的概率丢失数据

缺点：

官方对于ActiveMQ5.x的维护较少，高吞吐量使用场景较少

Kafka

使用率较高，大数据工程师的杀手锏，一般大数据内的消息传输绕不开Kafka，这款为大数据而生的消息中间件，以其百万级TPS吞吐量而闻名，在数据采集，传输，存储中发挥着重要的作用。

优点：

性能卓越，单机写入TPS在百万条/秒，最大的有点就是吞吐量高，时效性ms，可用性非常高，Kafka是分布式的，一个数据多个副本，少数宕机不会丢失数据，不会导致不可用，消费者采用Pull方式获取消息，消息有序，通过控制能够保证消息仅被消费者使用一次；有优秀的Kafka WEB管理页面Kafka-Manager；在日志领域比较成熟，功能较为简单，主要支持简单的MQ功能，在大数据领域的实时计算和日志采集大规模使用。

缺点：

单机超过64个队列/分区，Load会发生明显的飚高现象(CPU飚高)，队列越多，load越高，消息响应时间就会变长，使用短轮询方式，时效性取决于短轮询间隔时间，消费失败不支持重试，支持消息顺序，但是一台代理宕机后会出现消息乱序，社区更新较慢。

RockerMQ

出自阿里巴巴的开源产品，参考了Kafka，并做出了一些改进，被阿里巴巴广泛应用在订单，交易，充值，流计算，消息推送，日志，binlog分发等场景。

优点：

单机吞吐量10万级，分布式架构，可用性很高，可以做做到消息0丢失，MQ功能较为完善，扩展性好，支持10亿级别的消息堆积，不会因为消息堆积而造成性能下降，源码是java可以自己阅读并定制自己公司的MQ。

缺点：

支持客户端较少，目前主要是java，c++，c++目前不太成熟。没有在MQ核心去实现JMS等接口，有些系统需要迁移大量代码。

RabbitMQ：

2007年发布，是一个在AMQP（高级消息队列协议）的基础上完成的，可复用的企业消息队列系统，是当前最主流的消息中间件之一。

优点：

由于erlan语言的高并发特性，性能较好，吞吐量万级，MQ功能比较完备，健壮，稳定，医用，跨平台，支持多种语言（Python，ruby，.net，JAVA，JMS，C,PHP,ActionScript,XMPP，STMP等）支持AJAX文档齐全，开源提供管理界面非常棒，用起来很好用，社区活跃度高，更新频率高

缺点：商业版需要收费，学习成本较高。

官网：www.rabbitmq.com/news.html

1.4.MQ选择

1.Kafka：适合产生大量数据的互联网服务的数据收集业务，大型公司可以选用，如果有日志采集功能首选Kafka。

2.RocketMQ：为金融互联网而生，对于可靠性要求很高的场景，以及业务削峰，如果类似于阿里的双十一这样的高并发场景，建议选用。

3.RabbitMQ：结合elang语言本身特点，性能好时效性微秒级，社区活跃度很高，管理界面用起来十分方便，如果数据量没有那么大，中小型公司有效选的RabbitMQ。

2.RabbitMQ

2.1.RabbitMQ的概念

RabbitMQ是一个消息中间件；他接受并转发消息，可以把他当做一个快递站点，当你发送一个包裹时，你把你的包裹放到快递点，快递员会把你的快递送到收件人那里，他和快递站最大的区别在于，他不处理快件，而是接收，存储，转发消息数据。

2.2.四大核心概念

生产者：产生数据发送消息的程序是生产者

交换机：他一方面接受来自生产者的消息，另一方面将消息推送到队列中，交换机必须确切的知道如何处理它接收道德消息，试讲这些消息推送到特定的队列还是多个队列，亦或者丢弃，这都由交换机类型决定。（一个交换机可以对应一个队列也可以对应多个队列）

队列：他只能将消息储存在队列中，仅受到主机和磁盘限制的约束，本质上是一个消息称缓存区，许多生产者可以将消息发送到一个队列，许多消费者也可以从一个队列接收数据。

消费者：消费与接收的含义相似，大多数时候消费者是一个等待接收消息的程序，请注意生产者，消费者和消息中间件很多时候不在同一个机器上，同一个应用程序既可以是生产者也可以是消费者。

2.3.六大核心部分（六大模式）

2.3.1 Hello Word（简单模式）

2.3.2 Work Queues（工作模式）

2.3.3 Publish/Subscribe（发布订阅模式）

2.3.4 Routing（路由模式）

2.3.5 Topics （主题模式）

2.3.6 Publicher Confirms（发布确认模式） 

2.4.RabbitMQ名词关系

Producer：生产者

Broker：接收和转发消息，RabbitMQ Server 就是Message Broker，包含交换机和队列

Connection：连接，每一个生产者，消费者和Broker会建立一个链接。而每一个连接中会有多个信道（Channel），创建链接消耗比较大，所以只建立一次链接，但是有多个信道，每次发消息只占用一个信道。

2.5.安装

MQ安装步骤

官网：www.rabbitmq.com/download.html

1.升级所有包同时也升级软件和系统内核

yum -y update

  • yum -y update 升级所有包同时也升级软件和系统内核yum -y upgrade

  • 只升级所有包，不升级软件和系统内核

2.安装EPEL YUM源

yum -y install epel-release

3. 安装Erlang 环境

yum -y install erlang socat

4.查看版本

erl -version

5. 我们这时候就需要去看看官网上 Erlang 对应版本的 RabbitMQ 需要我们下载什么版本。

查看版本对应

6.根据系统版本Elang环境版本下班相应的MQ，然后去安装一下

查寻MQ版本

7.下载成功之后上传到Linux中运行

rpm -Uvh rabbitmq-server-3.7.26-1.el8.noarch.rpm

8.启动MQ

systemctl start rabbitmq-server

9.查看状态

systemctl status rabbitmq-server

其他命令

设置开机自启动

 chkconfig rabbitmq-server on

启动命令

/sbin/service rabbitmq-server start

查看服务状态

/sbin/service rabbitmq-server status

 

停止服务

 chkconfig rabbitmq-server stop

MQ后台管理安装

1.关闭MQ进程，查看状态

 2.安装插件

rabbitmq-plugins enable rabbitmq_management

3.重新运行

​chkconfig rabbitmq-server start

4.访问域名+15672，发现并不能访问，因为防火墙未关闭，关闭防火墙

systemctl stop firewalld

开机不启动防火墙

systemctl enable firewalld

开放端口

sudo firewall-cmd --add-port=15672/tcp --permanent

 重启防火墙

firewall-cmd --reload

 查看开放端口号

firewall-cmd --list-all

5.访问地址

默认账号密码为:guest guest

提示没有权限，不能登陆

6.添加一个新的账户

 rabbitmqctl add_user admin 123

7.设置账户权限

超级管理员

rabbitmqctl set_user_tags admin adminsrator

设置读写权限

 rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

8.查看全部用户权限

rabbitmqctl list_users

3.JAVA集成

3.1.Hello Word （简单模式）

3.1.1.创建maven项目，并引入依赖

  1.             <plugin>
  2. 				<groupId>org.springframework.boot</groupId>
  3. 				<artifactId>spring-boot-maven-plugin</artifactId>
  4. 				<configuration>
  5. 					<excludes>
  6. 						<exclude>
  7. 							<groupId>org.projectlombok</groupId>
  8. 							<artifactId>lombok</artifactId>
  9. 						</exclude>
  10. 					</excludes>
  11. 				</configuration>
  12. 			</plugin>
  13.  
  14. 	    <dependency>
  15. 			<groupId>com.rabbitmq</groupId>
  16. 			<artifactId>amqp-client</artifactId>
  17. 			<version>5.8.0</version>
  18. 		</dependency>
  19. 			<dependency>
  20. 			<groupId>commons-io</groupId>
  21. 			<artifactId>commons-io</artifactId>
  22. 			<version>2.6</version>
  23. 		</dependency>

3.1.2.创建生产者

  1. package com.rabbitmq.one;
  2.  
  3.  
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.Connection;
  6. import com.rabbitmq.client.ConnectionFactory;
  7.  
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  10.  
  11. public class Producer {
  12.     //队列名称
  13.     public static final String QUEUE_NAME="hello";
  14.  
  15.     public static void main(String[] args) throws IOException, TimeoutException {
  16.         //创建一个连接工厂
  17.         ConnectionFactory factory=new ConnectionFactory();
  18.         //设置工厂ip 连接队列
  19.         factory.setHost("112.124.34.53");
  20.         //设置用户名,密码
  21.         factory.setUsername("admin");
  22.         factory.setPassword("123");
  23.  
  24.         //创建链接
  25.         Connection connection = factory.newConnection();
  26.         //获取连接中的信道
  27.         Channel channel = connection.createChannel();
  28.         /**
  29.          *  生成一个队列
  30.          *  参数：
  31.          *  1.队列名称
  32.          *  2.是否持久化（默认在内存中，是否持久化到磁盘上）
  33.          *  3.是否消息共享（该队列是否只供一个消费者消费）
  34.          *  4.是否自动删除（该队列断开连接以后，是否自动删除）
  35.          *  5.其它参数（例如：延时消息，死信消息）
  36.          */
  37.         channel.queueDeclare(QUEUE_NAME,false,false,false,null);
  38.         String massage="hello word";
  39.         /**
  40.          *  发送消息
  41.          *  参数:
  42.          *  1.发送到哪个交换机
  43.          *  2.路由的key（本次为队列名）
  44.          *  3.其它参数
  45.          *  4.发送的消息
  46.          */
  47.         channel.basicPublish("",QUEUE_NAME,null, massage.getBytes());
  48.         System.out.println("消息已发送完毕");
  49.  
  50.     }
  51.  
  52. }
  53.

 有三个准备完毕的消息。

3.1.3.创建消费者

  1. package com.rabbitmq.one;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6. import java.util.concurrent.TimeoutException;
  7.  
  8. /**
  9.  * @Description
  10.  * 消费者接受消息
  11.  * @ClassName Consumer
  12.  * @Author LY
  13.  * @Date 2023/10/30 10:31
  14.  **/
  15.  
  16. public class Consumer {
  17.     //队列名称
  18.     public static final String QUEUE_NAME="hello";
  19.  
  20.     public static void main(String[] args) throws IOException, TimeoutException {
  21.         //创建一个连接工厂
  22.         ConnectionFactory factory=new ConnectionFactory();
  23.         //设置工厂ip 连接队列
  24.         factory.setHost("112.124.34.53");
  25.         //设置用户名,密码
  26.         factory.setUsername("admin");
  27.         factory.setPassword("123");
  28.         //创建链接
  29.         Connection connection = factory.newConnection();
  30.         //获取连接中的信道
  31.         Channel channel = connection.createChannel();
  32.         /**
  33.          *  消费消息
  34.          *  参数:
  35.          *  1.消费哪个队列
  36.          *  2.消费成功是否自动应答
  37.          *  3.消费者未成功消费的回调（接口）
  38.          *  4.消费者取消消费的回调（接口）
  39.          */
  40.         //声明参数3 接收消息
  41.         DeliverCallback deliverCallback=(consumerTag,massage)->{
  42.             System.out.println(new String(massage.getBody()));
  43.         };
  44.         //声明参数4 取消消息接收时的回调
  45.         CancelCallback cancelCallback=consumerTag->{
  46.             System.out.println("消息接收被中断");
  47.         };
  48.  
  49.         channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
  50.     }
  51. }

3.2.Work Queues（工作模式）

该模式主要为了避免立即执行资源密集型任务，而不得不等待他的完成。相反我们安排任务在之后执行，我们把任务作为消息发送到队列内，在后台运行的工作进程将弹出任务并最终执行作业，当有多个线程时，这些工作线程将一起处理这些任务。

3.21.轮训分发

一个生产者发送消息，多个工作线程接收，工作线程为竞争关系且一个消息只会被处理一次。

3.2.2.抽取工具类

  1. package com.rabbitmq.config;
  2.  
  3. public  class  RabbitMQConfig {
  4.     public static final String HOST="112.124.34.53";
  5.     public static final String USERNAME="admin";
  6.     public static final String PASSWORD="123";
  7. }
  1. package com.rabbitmq.utils;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6.  
  7. import java.io.IOException;
  8. import java.util.concurrent.TimeoutException;
  9. /**
  10.  * @Description
  11.  * 获取连接工具类
  12.  * @ClassName RabbitMQUtil
  13.  * @Author LY
  14.  * @Date 2023/10/30 15:49
  15.  **/
  16.  
  17. public class RabbitMQUtil {
  18.     //获得一个链接
  19.     public static Channel getChannel(String host, String userName, String passWord) throws IOException, TimeoutException {
  20.         //新建一个工厂
  21.         ConnectionFactory factory=new ConnectionFactory();
  22.         //设置参数
  23.         factory.setHost(host);
  24.         factory.setUsername(userName);
  25.         factory.setPassword(passWord);
  26.         //创建连接
  27.         Connection connection = factory.newConnection();
  28.         //创建信道
  29.         Channel channel = connection.createChannel();
  30.         return channel;
  31.     }
  32. }

3.2.3.消费者代码

  1. package com.rabbitmq.two;
  2.  
  3. import com.rabbitmq.client.CancelCallback;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rabbitmq.client.Delivery;
  7. import com.rabbitmq.config.RabbitMQConfig;
  8. import com.rabbitmq.utils.RabbitMQUtil;
  9.  
  10. import java.io.IOException;
  11. import java.util.concurrent.TimeoutException;
  12.  
  13. public class Worker {
  14.     //队列名称
  15.     public static final String QUEUE_NAME="hello";
  16.  
  17.  
  18. }
  19.  
  20.  
  21. class Worker01{
  22.     public static void main(String[] args) throws IOException, TimeoutException {
  23.         Channel channel = RabbitMQUtil.getChannel();
  24.         System.out.println("Worker01等到接受消息");
  25.         //接收消息
  26.         channel.basicConsume(Worker.QUEUE_NAME,true,new MyDeliverCallback(),new MyCancelCallback());
  27.     }
  28. }
  29.  
  30.  
  31. //消费者接收消息参数3
  32. class MyDeliverCallback implements DeliverCallback {
  33.     @Override
  34.     public void handle(String s, Delivery delivery) throws IOException {
  35.         System.out.println("接收到的消息：" + new String(delivery.getBody()));
  36.     }
  37. }
  38. //消费者接收消息参数4
  39. class MyCancelCallback implements CancelCallback {
  40.     @Override
  41.     public void handle(String s) throws IOException {
  42.         System.out.println("消息接收被中断");
  43.     }
  44. }

3.2.4.生产者代码

  1. package com.rabbitmq.two;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.utils.RabbitMQUtil;
  5.  
  6. import java.io.IOException;
  7. import java.util.concurrent.TimeoutException;
  8.  
  9. public class Producer {
  10.     public static final String QUEUE_NAME="hello";
  11.     public static void main(String[] args) throws IOException, TimeoutException {
  12.         Channel channel = RabbitMQUtil.getChannel();
  13.         channel.queueDeclare(QUEUE_NAME,false,false,false,null);
  14.         for (int i = 1; i <= 20; i++) {
  15.             String massage="hello word"+i;
  16.             channel.basicPublish("",QUEUE_NAME,null, massage.getBytes());
  17.             System.out.println("消息已发送完毕"+massage);
  18.         }
  19.  
  20.     }
  21. }

3.2.5.结果：

消息会轮流发送给work01,work02,work01,work02...

3.3.消息应答

3.3.1.概念：

如果有一个逻辑比较复杂，处理时间比较长，其中某个线程工作只进行了一部分然后宕机了，导致任务并没有完成，然后队列中的消息又被删除了，那么就意味着消息丢失了。

为了防止消息丢失，RabbitMQ引入了消息应答，消费者在接收并处理消息完成之后，消费者会通知MQ，此时MQ才会删除消息。

自动应答：

这种模式需要在高吞吐量和数据传输安全性方面做权衡，没有对消息的数量进行限制，所以这种模式仅是用于消费者可以高效并以某种速率能够处理这些消息的情况下使用。

手动应答：

A.Channel.basicAck（用于肯定应答）：

表示MQ已经知道该消息并且成功的处理消息，可以将其丢弃了。

B.Channel.basicNack（用于否定确认）

C.Channel.basicReject（用于否定确认）：与Channel.basicNack相比少一个参数（是否批量处理），表示不处理该消息了，直接拒绝，可以将其丢弃。

3.3.2.Multiple的解释

手动应答的好处是可以批量处理，减少网络拥堵。

例如：Channel.basicAck（8,multiple）,如果multiple为true的话他会处理此信道上所有的未被确认的应答消息，而如果是false则只处理8。

3.3.3.消息自动重新入队

如果由于某些原因失去连接（其通道已关闭，链接已关闭或TCP链接丢失），导致消息未发送ACK确认，RabbitMQ将了解到消息未完全处理，并将其重新排队，如果此时其他消费者可以处理，他将会很快的将其重新分发给另一个消费者，这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

3.3.4.消息手动应答代码

  1. package com.rabbitmq.three;
  2.  
  3. import com.rabbitmq.backInterface.MyCancelCallback;
  4. import com.rabbitmq.backInterface.MyDeliverCallback;
  5. import com.rabbitmq.client.Channel;
  6. import com.rabbitmq.client.DeliverCallback;
  7. import com.rabbitmq.client.Delivery;
  8. import com.rabbitmq.config.RabbitMQConfig;
  9. import com.rabbitmq.utils.RabbitMQUtil;
  10.  
  11. import java.io.IOException;
  12. import java.util.Scanner;
  13. import java.util.concurrent.TimeoutException;
  14.  
  15. public class Main {
  16.  
  17. }
  18. /**
  19.  * @Description
  20.  * 生产者发送消息
  21.  * @ClassName Producer
  22.  * @Author LY
  23.  * @Date 2023/10/31 14:00
  24.  **/
  25.  
  26. class Producer{
  27.     public static void main(String[] args) throws IOException, TimeoutException {
  28.         Channel channel = RabbitMQUtil.getChannel();
  29.         //声明一个队列
  30.         channel.queueDeclare(RabbitMQConfig.TASK_ACK_QUEUE_NAME,false,false,false,null);
  31.         //输入信息
  32.         Scanner scanner=new Scanner(System.in);
  33.         while (scanner.hasNext()){
  34.             String msg = scanner.next();
  35.             channel.basicPublish("",RabbitMQConfig.TASK_ACK_QUEUE_NAME,null,msg.getBytes("UTF-8"));
  36.             System.out.println("生产者发出消息："+msg);
  37.         }
  38.  
  39.     }
  40. }
  41. /**
  42.  * @Description
  43.  * 消费者缓慢处理消息(10秒)
  44.  * @ClassName Consumer01
  45.  * @Author LY
  46.  * @Date 2023/10/31 14:00
  47.  **/
  48.  
  49. class Consumer01{
  50.     public static void main(String[] args) throws IOException, TimeoutException {
  51.         Channel channel = RabbitMQUtil.getChannel();
  52.         System.out.println("Consumer01接收消息，沉睡50秒");
  53.         DeliverCallback deliverCallback=(s,delivery) ->{
  54.                 try {
  55.                     Thread.sleep(50000);
  56.                 } catch (InterruptedException e) {
  57.                     throw new RuntimeException(e);
  58.                 }
  59.                 System.out.println("接收到的消息：" + new String(delivery.getBody()));
  60.             /**
  61.              * 手动应答
  62.              * 参数：
  63.              * 1.消息标记 tag delivery.getEnvelope().getDeliveryTag()
  64.              * 2.是否批量应答
  65.              */
  66.             channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
  67.         };
  68.         Boolean autoAsk=false;
  69.         //采用手动应答
  70.         channel.basicConsume(RabbitMQConfig.TASK_ACK_QUEUE_NAME,autoAsk,deliverCallback,new MyCancelCallback());
  71.     }
  72. }
  73.  
  74. /**
  75.  * @Description
  76.  * 消费者快速处理消息
  77.  * @ClassName Consumer01
  78.  * @Author LY
  79.  * @Date 2023/10/31 14:00
  80.  **/
  81.  
  82. class Consumer02{
  83.     public static void main(String[] args) throws IOException, TimeoutException {
  84.         Channel channel = RabbitMQUtil.getChannel();
  85.         System.out.println("Consumer02接收消息");
  86.         Boolean autoAsk=false;
  87.         //采用手动应答
  88.         channel.basicConsume(RabbitMQConfig.TASK_ACK_QUEUE_NAME,autoAsk,new MyDeliverCallback(),new MyCancelCallback());
  89.     }
  90.  
  91. }

使用一个Thread.sleep(50000)来模拟一个复杂的业务逻辑，当我们对生产者输入消息AA时，第一个Consumer01会在50秒后输出小心并应答，此时该消息会被MQ销毁，Consumer02模拟一个简单的业务逻辑，当我们发送第二个消息BB会被Consumer02瞬间处理，此时我们发送CC，当消息发送成功后，我们关闭Consumer01，此时模拟系统意外宕机等突发情况，而Consumer01关闭后，Consumer02则会处理我们发送的消息CC被Consumer01接收到后，处理消息结束之前并未被销毁，这样可以有效防止各种突发状况导致的数据丢失。

3.4.RabbitMQ持久化

3.4.1.概念

当某个系统意外关闭或宕机时可以使用手动应答来防止数据丢失，而如果有意外情况导致RabbitMQ意外关闭或，他会忽视队列和消息，除非告知他不要这样做，确保消息不会丢失需要做两件事：队列持久化，消息持久化。

3.4.2.队列持久化

之前我们创建的队列都是非持久化的，MQ如果重启，该队列就会被删除掉，要实现队列持久化，需要在生命队列时把durable参数设置为持久化。

但是需要注意之前的队列并非持久化的，需要把原先的队列删除掉，或者重新创建一个持久换的队列，否则就会出现错误。

非持久化队列：

持久化队列：

3.4.3.消息持久化

要想让消息持久化需要再生产者发送消息时，设置第三个参数为：MessageProperties.PERSISTENT_TEXT_PLAIN

此参数为标记消息持久化，并且不能完全保证消息不会丢失，尽管他会标记消息保存到磁盘，但是他在消息写入磁盘之前，存在一个间隔点。此时并没有写入磁盘，持久性并不欠，但是对于简单队列而言已经绰绰有余了。如果需要更强有力的持久化策略，则需要配合发布确认模式使用。

3.4.4.不公平分发

MQ默认情况下才从轮训分发，但是在某种业务场景下，这种策略并非很好，比如两个消费者在处理任务，一个非常快，一个非常慢，这个时候继续采用轮训分发就会导致快的大部分时间都处于空闲状态，而慢的一直都在处理任务，这种情况下轮训分发其实并不太好，但是MQ并不知道会出现这种情况，在我们不指定的情况下，他依旧轮讯分发。

为了避免这种情况，我们可以设置参数channel.basicQos(1)

这个设置应该由消费者在接收消息之前来设置。

3.4.5.预期值

channel.basicQos(prefetch)，这个参数类似于权重（ngnix的weight参数），我们设置消费者A为5，消费者B为3，那么无论他们谁处理的快，谁处理的慢，8个消息一定会分发给消费者A5条，消费者B 3 条。

3.5.发布确认模式

1.必须设置队列必须持久化。

2.必须设置队列中的消息持久化。

3.发布确认。

生产者发送消息之后，MQ将信息保存到磁盘上之后，MQ会向生产者发送一个信息，消息已经被保存在磁盘上，这样可以完全保证，消息已经被MQ保存在磁盘上，消息完全不会丢失。

3.5.1.开启发布确认方法

再生产者中，发消息之前，创建信道之后，调用channel.confirmSelect();开启发布确认。

  1.  Channel channel = RabbitMQUtil.getChannel();
  2.  channel.confirmSelect();

3.5.2.单个发布确认

这是一种简单的确认方式，它是一种同步确认发布的方式，只有前一个消息确认发布后，才会继续发布，waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间内没有被确认，那么他将抛出异常。

这个方式有一个最大的缺点，发布速度特别慢，因为如果没有确认发布，就会阻塞所有的后续消息发布，这种方式最多提供每秒不超数百条的消息吞吐量，当然对于某些应用程序来讲，这已经足够了。

  1.  //单个确认
  2.     public static void  confirmOne() throws IOException, TimeoutException, InterruptedException {
  3.         Channel channel = RabbitMQUtil.getChannel();
  4.         String queueName= UUID.randomUUID().toString();
  5.         channel.queueDeclare(queueName,false,false,false,null);
  6.  
  7.         //开启发布确认
  8.         channel.confirmSelect();
  9.         long star=System.currentTimeMillis();
  10.         for (int i = 0; i < MESSAGE_COUNT; i++) {
  11.             String s=i+"";
  12.             channel.basicPublish("",queueName,null,s.getBytes());
  13.             boolean flag = channel.waitForConfirms();
  14.             if (flag){
  15.                 System.out.println("消息发送成功"+i);
  16.             }
  17.         }
  18.         long end=System.currentTimeMillis();
  19.  
  20.         long l = end - star;
  21.         System.out.println("发布"+MESSAGE_COUNT+"条单个确认耗时"+l);
  22.         //1.单个发布确认  发布1000条单个确认耗时38759ms
  23.     }

3.5.3.批量发布确认

与单个确认相比，先发布一批消息，然后以确认可以极大地提高吞吐量，当然这种方式的缺点就是，党发不出现问题是，不知道哪个消息出现了问题，我们必须将整个批处理保存在内存中，用以记录重要的信息二手重新发布，当然这种方式也是同步的，也一样阻塞线程。

  1.  //批量确认
  2.     public static void confirmBatch() throws IOException, TimeoutException, InterruptedException {
  3.         Channel channel = RabbitMQUtil.getChannel();
  4.         String queueName = UUID.randomUUID().toString();
  5.         channel.queueDeclare(queueName, false, false, false, null);
  6.  
  7.         //开启发布确认
  8.         channel.confirmSelect();
  9.         long star = System.currentTimeMillis();
  10.         //批量发布确认
  11.         //批量确认消息条数
  12.         Integer bathCount = 100;
  13.         for (int i = 1; i <= MESSAGE_COUNT; i++) {
  14.             String s = i + "";
  15.             channel.basicPublish("", queueName, null, s.getBytes());
  16.             if (i % bathCount == 0) {
  17.                 //每100条确认一次
  18.                 channel.waitForConfirms();
  19.                 System.out.println("消息发送成功" + i);
  20.             }
  21.         }
  22.         long end = System.currentTimeMillis();
  23.         long l = end - star;
  24.         System.out.println("发布" + MESSAGE_COUNT + "条每100条确认耗时" + l);
  25.         //发布1000条每100条确认耗时390
  26.     }

3.5.4.异步发布确认

相比于前两个同步确认，他的可靠性高，效率好，他利用回调函数来达到让消息的可靠传递的，这个中间件也是通过函数回调来确认消息是否投递成功。成功：确认应答，未成功：未确认应答。成功的不做处理，未收到的重新发送。因为是异步处理，所以生产者只需要发送数据，等待应答即可。由于是异步确认，所以会先发布成功，后续还会确认。

  1.  //异步确认
  2.     public static void asynConfirm() throws IOException, TimeoutException, InterruptedException {
  3.         Channel channel = RabbitMQUtil.getChannel();
  4.         String queueName = UUID.randomUUID().toString();
  5.         channel.queueDeclare(queueName, false, false, false, null);
  6.  
  7.         //开启发布确认
  8.         channel.confirmSelect();
  9.         long star = System.currentTimeMillis();
  10.         //异步发布确认
  11.         //准备监听器，监听发送成功和失败
  12.         channel.addConfirmListener(new MyConfirmCallbackSuccess(), new MyConfirmCallbackFail());
  13.         //批量确认消息条数
  14.         for (int i = 1; i <= MESSAGE_COUNT; i++) {
  15.             String s = i + "";
  16.             channel.basicPublish("", queueName, null, s.getBytes());
  17.         }
  18.         long end = System.currentTimeMillis();
  19.         long l = end - star;
  20.         System.out.println("发布" + MESSAGE_COUNT + "条异步确认耗时" + l);
  21.         //发布1000条异步确认耗时81
  22.     }

成功回调函数

  1. package com.rabbitmq.backInterface;
  2.  
  3. import com.rabbitmq.client.ConfirmCallback;
  4.  
  5. import java.io.IOException;
  6. //成功回调函数
  7. public class MyConfirmCallbackSuccess implements ConfirmCallback {
  8.     /**
  9.      * @Description
  10.      * @Author LY
  11.      * @Param [l, b]
  12.      * l:消息的标识，b是否批量确认
  13.      * @return void
  14.      * @Date 2023/11/1 10:09
  15.      **/
  16.  
  17.     @Override
  18.     public void handle(long l, boolean b) throws IOException {
  19.         System.out.println("确认的消息："+l);
  20.     }
  21. }

失败回调函数

  1. package com.rabbitmq.backInterface;
  2.  
  3. import com.rabbitmq.client.ConfirmCallback;
  4.  
  5. import java.io.IOException;
  6. //失败回调函数
  7. public class MyConfirmCallbackFail implements ConfirmCallback {
  8.     /**
  9.      * @Description
  10.      * @Author LY
  11.      * @Param [l, b]
  12.      * l:消息的标识，b是否批量确认
  13.      * @return void
  14.      * @Date 2023/11/1 10:09
  15.      **/
  16.     @Override
  17.     public void handle(long l, boolean b) throws IOException {
  18.         System.out.println("未确认的消息："+l);
  19.     }
  20. }

3.5.5.异步发布未确认消息

对于异步发布确认，消息发布未成功的消息应该有后续操作，无论是存储，还是重新发送，都要有一个处理方式。

最好的解决方案是吧未确认的消息放到一个基于内存的能被发布线程访问到的队列，比如说用ConcurrentLinkedQueue这个对俩在confirm callbacks与发布线程之间进行消息的传递。

这里需要三步：

1.消息发送之后记录所有已发送的消息

2.确认成功之后删除掉已确认的 消息

3.对未确认的消息进行特殊处理

  1.   //异步确认
  2.     public static void asynConfirm() throws IOException, TimeoutException, InterruptedException {
  3.         Channel channel = RabbitMQUtil.getChannel();
  4.         String queueName = UUID.randomUUID().toString();
  5.         channel.queueDeclare(queueName, false, false, false, null);
  6.  
  7.         //开启发布确认
  8.         channel.confirmSelect();
  9.         /**
  10.          * 线程安全有序地hashtable，是用于高并发的情况
  11.          * 1.轻松地将序号与消息进行关联
  12.          * 2.轻松批量的删除条数
  13.          * 3.支持高并发多线程
  14.          */
  15.         ConcurrentSkipListMap<Long, String> outstandingConfirms=new ConcurrentSkipListMap<>();
  16.         long star = System.currentTimeMillis();
  17.         //异步发布确认
  18.         //准备监听器，监听发送成功和失败
  19.         channel.addConfirmListener(new MyConfirmCallbackSuccess(outstandingConfirms), new MyConfirmCallbackFail(outstandingConfirms));
  20.         //批量确认消息条数
  21.         for (int i = 1; i <= MESSAGE_COUNT; i++) {
  22.             String s = i + "message";
  23.             channel.basicPublish("", queueName, null, s.getBytes());
  24.             //1.记录下所有发送的消息
  25.             outstandingConfirms.put(channel.getNextPublishSeqNo(),s );
  26.         }
  27.         long end = System.currentTimeMillis();
  28.         long l = end - star;
  29.         System.out.println("发布" + MESSAGE_COUNT + "条异步确认耗时" + l);
  30.         //发布1000条异步确认耗时81
  31.     }

成功处理:

  1. package com.rabbitmq.backInterface;
  2.  
  3. import com.rabbitmq.client.ConfirmCallback;
  4. import lombok.AllArgsConstructor;
  5. import lombok.Data;
  6. import lombok.NoArgsConstructor;
  7.  
  8. import java.util.Locale;
  9. import java.util.concurrent.ConcurrentNavigableMap;
  10. import java.util.concurrent.ConcurrentSkipListMap;
  11.  
  12. import java.io.IOException;
  13.  
  14. //成功回调函数
  15. @NoArgsConstructor
  16. @AllArgsConstructor
  17. @Data
  18. public class MyConfirmCallbackSuccess implements ConfirmCallback {
  19.     private ConcurrentSkipListMap concurrentSkipListMap;
  20.  
  21.  
  22.     /**
  23.      * @return void
  24.      * @Description
  25.      * @Author LY
  26.      * @Param [l, b]
  27.      * l:消息的标识，b是否批量确认
  28.      * @Date 2023/11/1 10:09
  29.      **/
  30.  
  31.     @Override
  32.     public void handle(long l, boolean b) throws IOException {
  33.         if (b) {
  34.             //如果是批量 清空
  35.             ConcurrentNavigableMap<Long, String> confirmed = concurrentSkipListMap.headMap(l);
  36.             confirmed.clear();
  37.         } else {
  38.             //非批量删除单个
  39.             concurrentSkipListMap.remove(l);
  40.         }
  41.         //2.删除确认的消息 剩余的是未确认的消息
  42.         System.out.println("确认的消息：" + concurrentSkipListMap.get(l)+"   消息标记："+l);
  43.     }
  44. }

未确认处理：

  1. package com.rabbitmq.backInterface;
  2.  
  3. import com.rabbitmq.client.ConfirmCallback;
  4. import lombok.AllArgsConstructor;
  5. import lombok.Data;
  6. import lombok.NoArgsConstructor;
  7.  
  8. import java.io.IOException;
  9. import java.util.concurrent.ConcurrentSkipListMap;
  10.  
  11. //失败回调函数
  12. @NoArgsConstructor
  13. @AllArgsConstructor
  14. @Data
  15. public class MyConfirmCallbackFail implements ConfirmCallback {
  16.    private ConcurrentSkipListMap concurrentSkipListMap;
  17.  
  18.     /**
  19.      * @Description
  20.      * @Author LY
  21.      * @Param [l, b]
  22.      * l:消息的标识，b是否批量确认
  23.      * @return void
  24.      * @Date 2023/11/1 10:09
  25.      **/
  26.     @Override
  27.     public void handle(long l, boolean b) throws IOException {
  28.         Object msg = concurrentSkipListMap.get(l);
  29.         System.out.println("未确认的消息："+msg+"未确认的消息标记是："+l);
  30.     }
  31. }

3.5.6.三种模式速度对比

  1.         //1.单个发布确认
  2.         confirmOne();//1.单个发布确认  发布1000条单个确认耗时38759ms
  3.         //2.批量发布确认
  4.         confirmBatch();//发布1000条每100条确认耗时390
  5.         //3.异步发布确认
  6.         asynConfirm();//发布1000条异步确认耗时81

单独发布消息：同步等待确认，简单，但是吞吐量有限

批量发布消息：批量同步等待确认，简单，合理的吞吐量，一旦出现问题很难具体到哪条消息。

异步处理：最佳的性能和资源使用，再出现错的情况下可以很好的控制，但实现起来稍难。

4交换机

在之前的模式中，我们创建了一个工作队列，假设工作队列的背后，每个人物都签好交付给一个消费者（工作进程）。在这一部分中，将做一些完全不同的事情，将消息传达给多个消费者，这种模式我们称之为：“发布/订阅模式”。

例如一个简单的系统日志。他由两个程序组成，第一个程序将发出日志消息，第二个程序是消费者，其中消费者将会启动两个，一个接收到后将消息存储在磁盘上，第二个消费者把消息打印在屏幕上，事实上第一个程序发出的等消息日志将官拨给所有消费者。

之前的模式，消费者都是竞争关系，同一个队里中的同一份消息只会被消费一次。

4.1.Exchanges

4.1.1.Exchanges概念

RabbitMQ消息传递模型的核心思想是：生产者生产的消息从不会直接发送到队列，就算是简单模式，我们也走的是默认交换机。实际上，通常生产者甚至都不知道这些消息传递带了那些队列中。想法生产者只能将消息发送到交换机（exchange），交换机的工作内容非常简单，一方面他接受来自生产者的消息，另一方面将他们推入队列。交换机必须确切知道如何处理收到的消息，是应该把这些消息放到特定的队列还是把他们放到许多队列中，或者应该丢弃他们，这就由交换机的类型来决定。

4.1.2.Exchange类型

直接（direct）（路由类型）

主题（topic）

标题（headers）（头类型）

扇出（fanout）（发布订阅类型）

无名exchange

4.1.3.无名exchange

之前的发送消息我们并没有制定交换机，之前之所以能实现将消息发送到队列中，因为我们使用的是默认交换机，通常用字符串（“”）进行标识。

channel.basicPublish("", queueName, null, s.getBytes());

第一个参数是交换机名称，空字符串表示默认或者无名交换机；消息之所以能发送到队列中其实是由routingKey()绑定key指定的。

4.2.临时队列

每当我们连接到RabbitMQ时，我们需要一个全新的空队列，为此我们可以创建一个具有随机名称的队列，或者让服务器为我们选择一个随机队列名称那就更好了，其次我们一旦断开连接，队列就会自动删除，队列没有被持久化，持久化标记没有。

String queueName=channel.queueDedare().getQueue;

4.3.绑定（bindings）

1.Add a new queue

2.添加一个交换机

3.交换机与队列绑定

4.通过123与hello1相连接

4.4.fanout（发布订阅）（扇出）

4.4.1.概念

这种类型非常简单，正如名称中那样，它是将受到的所有信息，广播到他知道的所有队列中，系统中默认有些exchange类型。

4.4.2.fanout代码

  1. package com.rabbitmq.five;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.DeliverCallback;
  5. import com.rabbitmq.client.Delivery;
  6. import com.rabbitmq.config.RabbitMQConfig;
  7. import com.rabbitmq.utils.RabbitMQUtil;
  8.  
  9. import java.io.IOException;
  10. import java.util.Scanner;
  11. import java.util.concurrent.TimeoutException;
  12.  
  13. public class Logs {
  14. }
  15.  
  16.  
  17. class EmitLogs{
  18.     public static void main(String[] args) throws IOException, TimeoutException {
  19.         Channel channel = RabbitMQUtil.getChannel();
  20.         //声明一个交换机
  21.         channel.exchangeDeclare(RabbitMQConfig.EXCHANGE_NAME,"fanout");
  22.         Scanner sc=new Scanner(System.in);
  23.         while (sc.hasNext()){
  24.             String msg=sc.next();
  25.             channel.basicPublish(RabbitMQConfig.EXCHANGE_NAME,"",null,msg.getBytes());
  26.             System.out.println("生产者发出消息："+msg);
  27.         }
  28.     }
  29. }
  30.  
  31. class ReceiveLogs01{
  32.     public static void main(String[] args) throws IOException, TimeoutException {
  33.         Channel channel = RabbitMQUtil.getChannel();
  34.         //声明一个交换机
  35.         channel.exchangeDeclare(RabbitMQConfig.EXCHANGE_NAME,"fanout");
  36.         //声明一个队列 临时队列 队列名称随机
  37.         String queue = channel.queueDeclare().getQueue();
  38.         //绑定交换机和队列
  39.         channel.queueBind(queue,RabbitMQConfig.EXCHANGE_NAME,"");
  40.         System.out.println("ReceiveLogs01等待接收消息，并将消息打印在控制台");
  41.         channel.basicConsume(queue, true, (s,delivery) ->{
  42.             System.out.println("ReceiveLogs01控制台打印："+new String(delivery.getBody()));
  43.             },(s)->{});
  44.  
  45.     }
  46. }
  47. class ReceiveLogs02{
  48.     public static void main(String[] args) throws IOException, TimeoutException {
  49.         Channel channel = RabbitMQUtil.getChannel();
  50.         //声明一个交换机
  51.         channel.exchangeDeclare(RabbitMQConfig.EXCHANGE_NAME,"fanout");
  52.         //声明一个队列 临时队列 队列名称随机
  53.         String queue = channel.queueDeclare().getQueue();
  54.         //绑定交换机和队列
  55.         channel.queueBind(queue,RabbitMQConfig.EXCHANGE_NAME,"");
  56.         System.out.println("ReceiveLogs02等待接收消息，并将消息打印在控制台");
  57.         channel.basicConsume(queue, true, (s,delivery) ->{
  58.             System.out.println("ReceiveLogs02控制台打印："+new String(delivery.getBody()));
  59.         },(s)->{});
  60.     }
  61. }

4.5.Direct exchange

直接交换机  路由模式

队列支队他绑定的交换机的消息感兴趣，绑定参数routingKey来表示也可以称该参数为binding key，创建绑定我们用代码channel.queueBind(queue,RabbitMQConfig.EXCHANGE_NAME,"routingKey");绑定之后的意义由交换类型决定。

当routhingKey相同时，就是发布订阅模式，当帮懂得routhingKey不同时就是路由模式，也叫直接交换机。

4.5.1 路由模式多重绑定

  1. package com.rabbitmq.six;
  2.  
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.config.RabbitMQConfig;
  6. import com.rabbitmq.utils.RabbitMQUtil;
  7.  
  8. import java.io.IOException;
  9. import java.util.ArrayList;
  10. import java.util.Scanner;
  11. import java.util.concurrent.TimeoutException;
  12.  
  13. public class directLogs {
  14. }
  15.  
  16. class EmitLogs{
  17.     public static void main(String[] args) throws IOException, TimeoutException {
  18.         int index=0;
  19.         String [] routingKey={"info","warning","error"};
  20.         Channel channel = RabbitMQUtil.getChannel();
  21.         //声明一个交换机
  22.         Scanner sc=new Scanner(System.in);
  23.         while (sc.hasNext()){
  24.             String msg=sc.next();
  25.             channel.basicPublish(RabbitMQConfig.DIRECT_EXCHANGE_NAME,routingKey[index%routingKey.length],null,msg.getBytes());
  26.             index++;
  27.             System.out.println("生产者发出消息："+msg);
  28.         }
  29.     }
  30. }
  31.  
  32. class ReceiveLogs01{
  33.     public static void main(String[] args) throws IOException, TimeoutException {
  34.         Channel channel = RabbitMQUtil.getChannel();
  35.         //声明一个交换机
  36.         channel.exchangeDeclare(RabbitMQConfig.DIRECT_EXCHANGE_NAME,BuiltinExchangeType.DIRECT);
  37.         //声明一个队列 临时队列 队列名称随机
  38.        channel.queueDeclare("console",false,false,false,null);
  39.         //绑定交换机和队列
  40.         channel.queueBind("console",RabbitMQConfig.DIRECT_EXCHANGE_NAME,"info");
  41.         channel.queueBind("console",RabbitMQConfig.DIRECT_EXCHANGE_NAME,"warning");
  42.         System.out.println("ReceiveLogs01等待接收消息，并将消息打印在控制台");
  43.         channel.basicConsume("console", true, (s,delivery) ->{
  44.             System.out.println("ReceiveLogs01Console控制台打印："+new String(delivery.getBody()));
  45.         },(s)->{});
  46.  
  47.     }
  48. }
  49. class ReceiveLogs02{
  50.     public static void main(String[] args) throws IOException, TimeoutException {
  51.         Channel channel = RabbitMQUtil.getChannel();
  52.         //声明一个交换机
  53.         channel.exchangeDeclare(RabbitMQConfig.DIRECT_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
  54.         //声明一个队列 临时队列 队列名称随机
  55.         channel.queueDeclare("disk",false,false,false,null);
  56.         //绑定交换机和队列
  57.         channel.queueBind("disk",RabbitMQConfig.DIRECT_EXCHANGE_NAME,"error");
  58.         System.out.println("ReceiveLogs02等待接收消息，并将消息打印在控制台");
  59.         channel.basicConsume("disk", true, (s,delivery) ->{
  60.             System.out.println("ReceiveLogs02控制台打印："+new String(delivery.getBody()));
  61.         },(s)->{});
  62.  
  63.     }
  64. }

此时第一次输入routingKey取到info，因为消费者1绑定的routingKey包含info，此时消费者1接收，第二次输入，routingKey取warning消费者1绑定的routingKey也包含warning，所以第二条数据也被消费者1接收，第三次输入routingKey取error，此时消息被消费者2接收。

4.6.Topic（主题交换机）

4.6.1概念

尽管direct交换机做了某些改进，但是他仍然有局限性，例如：日志类型有“info.base”,“info.advantage”，某个队列只需要“info.base”,那么这个时候direct交换机就做不到了。需要使用topic交换机。

4.6.2 topic要求

类型是topic交换机的信息的routing_key不能随便写，必须满足要求，他必须是一个单词表，以“.”分隔，这些单词可以是任意单词，例如“stock.usd.nyse”，“nyse.vmw”，“quick.orange.rabbit”，这种类型的，但是最大不能超过255字节。

在这个规则列表中，有两个替换符需要注意

*（星号）可以替代一个单词

#（井号）可以替代零个或多个单词

对于直接交换机来说，最多只能路由一个队列，可以捆绑多个，但是发送只会发送到一个队列中。

4.6.3 匹配案例

Q1=>绑定的是

        1.中间带orange带三个单词的字符串（*.orange.*）

Q2=>绑定的是

        1.最后一个单词是rabbit的三个单词（*.*.rabbit）

        2.第一个单词是lazy的多个单词（lazy.#）

      routing_key                             满足队列                         接收次数

quick.orange.rabbit                      Q1.1，Q2.1                        2

lazy.orange.elephant                   Q1.1                                    1

quick.orange.fox                          Q1.1                                    1

lazy.brown.fox                              Q2.2                                    1

lazy.pink.rabbit                             Q2.1，Q2.2                         1

quick.brown.fox                            无                                        0

quick.orange.male.rabbit              无                                        0 

lazy.orange.mel.rabbit                  Q2.2                                    1

注意：当一个队列绑定#，那么他将接收所有数据，类似于fanout，如果队列绑定键没有#和*那么该队列绑定类型就是direct。所以主题交换机包含了其他两个交换机。

4.6.4 Topic代码

  1. package com.rabbitmq.seven;
  2.  
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.config.RabbitMQConfig;
  6. import com.rabbitmq.utils.RabbitMQUtil;
  7.  
  8. import java.io.IOException;
  9. import java.util.concurrent.TimeoutException;
  10.  
  11. public class TopicExchange {
  12. }
  13. class EmitLogs{
  14.     public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
  15.  
  16.         String [] routingKey={"quick.orange.rabbit","lazy.orange.elephant"," quick.orange.fox","lazy.brown.fox","lazy.pink.rabbit","quick.brown.fox","quick.orange.male.rabbit","lazy.orange.mel.rabbit"};
  17.         Channel channel = RabbitMQUtil.getChannel();
  18.         //声明一个交换机
  19.         for (int i = 0; i < routingKey.length; i++) {
  20.             Thread.sleep(1500);
  21.             String msg=routingKey[i]+":"+i;
  22.             channel.basicPublish(RabbitMQConfig.TOPIC_EXCHANGE_NAME,routingKey[i],null,msg.getBytes());
  23.             System.out.println("生产者发出消息："+msg);
  24.         }
  25.  
  26.  
  27.     }
  28. }
  29. class ReceiveLogs01{
  30.     public static void main(String[] args) throws IOException, TimeoutException {
  31.         Channel channel = RabbitMQUtil.getChannel();
  32.         //声明一个交换机
  33.         channel.exchangeDeclare(RabbitMQConfig.TOPIC_EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
  34.         //声明一个队列 临时队列 队列名称随机
  35.         channel.queueDeclare("Q1",false,false,false,null);
  36.         //绑定交换机和队列
  37.         channel.queueBind("Q1",RabbitMQConfig.TOPIC_EXCHANGE_NAME,"*.orange.*");
  38.         System.out.println("ReceiveLogs01等待接收消息，并将消息打印在控制台");
  39.         channel.basicConsume("Q1", true, (s,delivery) ->{
  40.             System.out.println("接收队列：Q1绑定键："+delivery.getEnvelope().getRoutingKey()+"内容"+new String(delivery.getBody()));
  41.         },(s)->{});
  42.  
  43.     }
  44. }
  45. class ReceiveLogs02{
  46.     public static void main(String[] args) throws IOException, TimeoutException {
  47.         Channel channel = RabbitMQUtil.getChannel();
  48.         //声明一个交换机
  49.         channel.exchangeDeclare(RabbitMQConfig.TOPIC_EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
  50.         //声明一个队列 临时队列 队列名称随机
  51.         channel.queueDeclare("Q2",false,false,false,null);
  52.         //绑定交换机和队列
  53.         channel.queueBind("Q2",RabbitMQConfig.TOPIC_EXCHANGE_NAME,"*.*.rabbit");
  54.         channel.queueBind("Q2",RabbitMQConfig.TOPIC_EXCHANGE_NAME,"lazy.#");
  55.         System.out.println("ReceiveLogs02等待接收消息，并将消息打印在控制台");
  56.         channel.basicConsume("Q2", true, (s,delivery) ->{
  57.             System.out.println("接收队列：Q2绑定键："+delivery.getEnvelope().getRoutingKey()+"内容"+new String(delivery.getBody()));
  58.         },(s)->{});
  59.  
  60.     }
  61. }

生产者：

生产者发出消息：quick.orange.rabbit:0
生产者发出消息：lazy.orange.elephant:1
生产者发出消息： quick.orange.fox:2
生产者发出消息：lazy.brown.fox:3
生产者发出消息：lazy.pink.rabbit:4
生产者发出消息：quick.brown.fox:5
生产者发出消息：quick.orange.male.rabbit:6
生产者发出消息：lazy.orange.mel.rabbit:7

消费者1：

ReceiveLogs01等待接收消息，并将消息打印在控制台
接收队列：Q1绑定键：quick.orange.rabbit内容quick.orange.rabbit:0
接收队列：Q1绑定键：lazy.orange.elephant内容lazy.orange.elephant:1
接收队列：Q1绑定键： quick.orange.fox内容 quick.orange.fox:2

消费者2：

ReceiveLogs02等待接收消息，并将消息打印在控制台
接收队列：Q2绑定键：quick.orange.rabbit内容quick.orange.rabbit:0
接收队列：Q2绑定键：lazy.orange.elephant内容lazy.orange.elephant:1
接收队列：Q2绑定键：lazy.brown.fox内容lazy.brown.fox:3
接收队列：Q2绑定键：lazy.pink.rabbit内容lazy.pink.rabbit:4
接收队列：Q2绑定键：lazy.orange.mel.rabbit内容lazy.orange.mel.rabbit:7

4.7.死信队列

4.7.1概念

从概念上来讲，死信，指的是无法被消费的消息，一般来说producer将消息投递到broker或者直接到了queue中，consumer从queue中去除消息就行消费，但是某些时候由于特殊的原因，导致queue中的某些消息无法被消费，这样的消息如果没有后续处理就变成了死信，有私心自然就有了死信队列。

应用场景：为了保证订单业务的消息数据不丢失，需要使用到RabbitMQ中的死信队列机制，当消息发生异常时，将消息投入死信队列中，当系统恢复正常时，去除消费。还有比如说：用户在商城下单成功并单击支付后，在指定时间内尚未支付时自动失效。

4.7.2死信的来源

消息TTL过期（可以通过生产者设置，也可以通过消费者设置）

队列达到了最大长度（队列满了，无法在添加到MQ中）

消息被拒绝（basic.reject或者basic.nack）并且不放回队列中（requeue=false）

4.7.3死信代码

当设置TTL为10秒时，超过10秒的消息自动进入死信队列。

当设置最大长度为6时，队列内消息超过6条将进入死信队列。

当设置消息拒绝时，被拒绝的消息自动进入死信队列。

消费者

  1. package com.rabbitmq.eight;
  2.  
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.DeliverCallback;
  6. import com.rabbitmq.utils.RabbitMQUtil;
  7.  
  8. import java.io.IOException;
  9. import java.util.HashMap;
  10. import java.util.Map;
  11. import java.util.concurrent.TimeoutException;
  12.  
  13. public class Consumer01 {
  14.  
  15.     public static final String NORMAL_QUEUE = "normal_queue";
  16.     public static final String NORMAL_EXCHANGE = "normal_exchange";
  17.     public static final String DEAD_QUEUE = "dead_queue";
  18.     public static final String DEAD_EXCHANGE = "dead_exchange";
  19.  
  20.     public static void main(String[] args) throws IOException, TimeoutException {
  21.         Channel channel = RabbitMQUtil.getChannel();
  22.         // 声明普通和死信交换机
  23.         channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
  24.         channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
  25.  
  26.  
  27.         // 声明死信队列
  28.         channel.queueDeclare(DEAD_QUEUE, false, false, false, null);
  29.         // 死信的绑定
  30.         channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");
  31.  
  32.  
  33.         Map<String, Object> arguments = new HashMap<>();
  34.         // 普通队列设置对应的交换机
  35.         arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);
  36.         // 设置过期时间
  37.        // arguments.put("x-message-ttl", 100000);
  38.         // 设置死信队列的RouteKey
  39.         arguments.put("x-dead-letter-routing-key", "lisi");
  40.         // 设置队列最大长度
  41.        // arguments.put("x-max-length", 6);
  42.         // 声明普通队列
  43.         channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);
  44.         // 普通的绑定
  45.         channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");
  46.  
  47.         DeliverCallback deliverCallback = (consumerTag, message) -> {
  48.             String msg = new String(message.getBody());
  49.             //消息被拒绝
  50.             if (msg.equals("info5")) {
  51.                 System.out.println("Consumer01接收到消息" + message + "并拒绝签收该消息");
  52.                 channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
  53.             } else {
  54.                 System.out.println("consumer01接收到消息：" + msg);
  55.                 channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
  56.             }
  57.         };
  58.         channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, consumerTag -> {
  59.         });
  60.     }
  61. }
  62. class Consumer02 {
  63.  
  64.     public static final String NORMAL_QUEUE = "normal_queue";
  65.     public static final String NORMAL_EXCHANGE = "normal_exchange";
  66.     public static final String DEAD_QUEUE = "dead_queue";
  67.     public static final String DEAD_EXCHANGE = "dead_exchange";
  68.  
  69.     public static void main(String[] args) throws IOException, TimeoutException {
  70.         Channel channel = RabbitMQUtil.getChannel();
  71.         System.out.println("Consumer02等待接受消息：");
  72.         channel.basicConsume(DEAD_QUEUE, false, (s,d)->{
  73.             String msg = new String(d.getBody());
  74.             System.out.println("consumer01接收到消息：" + msg);
  75.             channel.basicAck(d.getEnvelope().getDeliveryTag(), false);
  76.         }, consumerTag -> {
  77.         });    }
  78. }

生产者

  1. package com.rabbitmq.eight;
  2.  
  3. import com.rabbitmq.client.AMQP;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.utils.RabbitMQUtil;
  6.  
  7. import java.io.IOException;
  8. import java.util.concurrent.TimeoutException;
  9.  
  10. public class Producer {
  11.  
  12.     public static final String NORMAL_EXCHANGE = "normal_exchange";
  13.  
  14.     public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
  15.         Channel channel = RabbitMQUtil.getChannel();
  16.         //设置过期时间
  17.        // AMQP.BasicProperties properties=new AMQP.BasicProperties().builder().expiration("10000").build();
  18.         for (int i = 0; i < 10; i++) {
  19.             Thread.sleep(1000);
  20.             String message = "info" + i;
  21.             channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", null, message.getBytes());
  22.             System.out.println("发送消息"+message);
  23.         }
  24.  
  25.     }
  26. }

4.8.延迟队列

当死信队列中消费者1永久消失，过期时间设置为10s，那么从生产者1到消费者2所花费的时间就是10s，延迟队列就是死信队列中的消息过期这一种情况。

4.8.1 概念

延时队列，队列内部是有虚的，最重要的特性就是体现在他的延时属性上，延时队列中的元素是希望在指定时间到了以后或者之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

4.8.2 使用场景

1.订单在十分钟内未支付自动取消。

2.新创建的店铺，如果10天内没有上传商品，则自动发送消息提醒。

3.用户注册成功后，如果三天内没有登陆则进行短信提醒。

4.用户发起退款，三天内没有得到处理则通知相关人员。

5.预定会议后，需要在预定的时间点前十分钟通知各个参会人员参加会议。

这些场景都有一个特点，需要在某个事件发生之后或者之前的指定时间点完成某一项任务，如：发生订单生成时间，在十分钟之后检查该订单支付状态，然后将未支付的订单进行关闭；看起来似乎是使用定时任务轮训查询数据，每秒一次，再取出需要被处理的数据。但是如果这种方式面对的试一下对于时间不是严格限制而是宽松意义上的某段时间，那么每天晚上拍个定时任务进行自动结算也是可行的，但是面对数据量比较大，并且时效性较强的场景，如：短期内未支付订单可能达到百万甚至是千万级别，对于如此庞大的数据，人就是用轮训的方式显然是不可取的，因为同一秒内无法完成所有订单的检查，同时给数据库带来很大的严厉，无法满足业务需求而且性能低下。

4.9.整合Spring Boot

4.9.1 新建Springboot项目

4.9.1.1 引入依赖
  1.  <dependencies>
  2.         <dependency>
  3.             <groupId>org.springframework.boot</groupId>
  4.             <artifactId>spring-boot-starter</artifactId>
  5.         </dependency>
  6.  
  7.         <dependency>
  8.             <groupId>org.springframework.boot</groupId>
  9.             <artifactId>spring-boot-starter-test</artifactId>
  10.             <scope>test</scope>
  11.         </dependency>
  12.         <dependency>
  13.             <groupId>org.springframework.boot</groupId>
  14.             <artifactId>spring-boot-starter-amqp</artifactId>
  15.         </dependency>
  16.         <dependency>
  17.             <groupId>com.alibaba</groupId>
  18.             <artifactId>fastjson</artifactId>
  19.             <version>1.2.47</version>
  20.         </dependency>
  21.         <dependency>
  22.             <groupId>org.springframework.boot</groupId>
  23.             <artifactId>spring-boot-starter-web</artifactId>
  24.         </dependency>
  25.         <dependency>
  26.             <groupId>org.projectlombok</groupId>
  27.             <artifactId>lombok</artifactId>
  28.         </dependency>
  29.         <dependency>
  30.             <groupId>io.springfox</groupId>
  31.             <artifactId>springfox-swagger2</artifactId>
  32.             <version>2.9.2</version>
  33.         </dependency>
  34.         <dependency>
  35.             <groupId>org.springframework.amqp</groupId>
  36.             <artifactId>spring-rabbit-test</artifactId>
  37.             <scope>test</scope>
  38.         </dependency>
  39.     </dependencies>
4.9.1.2修改配置文件
  1. spring.rabbitmq.host=112.124.34.53
  2. spring.rabbitmq.port=5672
  3. spring.rabbitmq.username=admin
  4. spring.rabbitmq.password=123
4.9.1.3添加swagger配置类
  1. package com.rabbitmq02.config;
  2.  
  3.  
  4. import org.springframework.amqp.core.*;
  5. import org.springframework.beans.factory.annotation.Qualifier;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  8.  
  9. import java.util.HashMap;
  10. import java.util.Map;
  11.  
  12. /**
  13.  * @Description
  14.  * @ClassName TtlQueueConfig
  15.  * @Author LY
  16.  * @Date 2023/11/6 14:24
  17.  **/
  18. @Configuration
  19. public class TtlQueueConfig {
  20.     //普通交换机名称
  21.     public static final String X_EXCHANGE="X";
  22.     //死信交换机名称
  23.     public static final String Y_DEAD_EXCHANGE="Y";
  24.     //普通队列名称
  25.     public static final String QUEUE_A="QA";
  26.     public static final String QUEUE_B="QB";
  27.     //死信队列名称
  28.     public static final String DEAD_QUEUE_D="QD";
  29.  
  30.     //声明直接交换机X
  31.     @Bean("xExchange")
  32.     public DirectExchange xExchange(){
  33.         return new DirectExchange(X_EXCHANGE);
  34.     }
  35.     //声明死信交换机Y
  36.     @Bean("yExchange")
  37.     public DirectExchange yExchange(){
  38.         return new DirectExchange(Y_DEAD_EXCHANGE);
  39.     }
  40.     //声明称普通队列QA
  41.     @Bean("queueA")
  42.     public Queue queueA(){
  43.         Map<String, Object> arg=new HashMap<>(3);
  44.         //死信交换机
  45.         arg.put("x-dead-letter-exchange",Y_DEAD_EXCHANGE);
  46.         //死信RoutingKey
  47.         arg.put("x-dead-letter-routing-key","YD");
  48.         //过期时间TTL
  49.         arg.put("x-message-ttl",10000);
  50.         return QueueBuilder.durable(QUEUE_A).withArguments(arg).build();
  51.     }
  52.     //声明称普通队列QB
  53.     @Bean("queueB")
  54.     public Queue queueB(){
  55.         Map<String, Object> arg=new HashMap<>(3);
  56.         //死信交换机
  57.         arg.put("x-dead-letter-exchange",Y_DEAD_EXCHANGE);
  58.         //死信RoutingKey
  59.         arg.put("x-dead-letter-routing-key","YD");
  60.         //过期时间TTL
  61.         arg.put("x-message-ttl",40000);
  62.         return QueueBuilder.durable(QUEUE_B).withArguments(arg).build();
  63.     }
  64.     //声明称死信队列QD
  65.     @Bean("queueD")
  66.     public Queue queueD(){
  67.         return QueueBuilder.durable(DEAD_QUEUE_D).build();
  68.     }
  69.     //绑定queueA和xExchange
  70.     @Bean
  71.     public Binding queueABindingX(@Qualifier("queueA") Queue queueA, @Qualifier("xExchange") DirectExchange
  72.             xExchange){
  73.         return BindingBuilder.bind(queueA).to(xExchange).with("XA");
  74.     }
  75.     //绑定queueB和xExchange
  76.     @Bean
  77.     public Binding queueBBindingX(@Qualifier("queueB") Queue queueB, @Qualifier("xExchange") DirectExchange
  78.             xExchange){
  79.         return BindingBuilder.bind(queueB).to(xExchange).with("XB");
  80.     }
  81.      //绑定queueD和yExchange
  82.     @Bean
  83.     public Binding queueDBindingy(@Qualifier("queueD") Queue queueD, @Qualifier("yExchange") DirectExchange
  84.             yExchange){
  85.         return BindingBuilder.bind(queueD).to(yExchange).with("YD");
  86.     }
  87.  
  88. }
4.9.1.4 新增生产者
  1. package com.rabbitmq02.controller;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.web.bind.annotation.GetMapping;
  7. import org.springframework.web.bind.annotation.PathVariable;
  8. import org.springframework.web.bind.annotation.RequestMapping;
  9. import org.springframework.web.bind.annotation.RestController;
  10.  
  11. import java.util.Date;
  12.  
  13. /***
  14.  * @Description
  15.  * 发从延迟消息
  16.  * @ClassName SendMessageController
  17.  * @Author LY
  18.  * @Date 2023/11/6 15:06
  19.  **/
  20. @Slf4j
  21. @RestController
  22. @RequestMapping("/ttl")
  23. public class SendMessageController {
  24.  
  25.     @Autowired
  26.    private RabbitTemplate rabbitTemplate;
  27.     //开始发消息
  28.     @GetMapping("/sendMsg/{message}")
  29.     public void sendMsg(@PathVariable String message){
  30.         log.info("当前时间={}，发送一条信息给两个ttl队列：{}",new Date().toString(),message);
  31.         rabbitTemplate.convertAndSend("X","XA","消息来自ttl为10s的队列："+new String(message.getBytes()) );
  32.         rabbitTemplate.convertAndSend("X","XB","消息来自ttl为40s的队列："+new String(message.getBytes()) );
  33.     }
  34.  
  35. }
4.9.1.5新增消费者
  1. package com.rabbitmq02.consumer;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.amqp.core.Message;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.stereotype.Component;
  8.  
  9. import java.util.Date;
  10.  
  11. /***
  12.  * @Description
  13.  * TTL消费者
  14.  * @ClassName DeadLetterQueueConsumer
  15.  * @Author LY
  16.  * @Date 2023/11/6 15:39
  17.  **/
  18. @Component
  19. @Slf4j
  20. public class DeadLetterQueueConsumer {
  21.     //接收消息
  22.     @RabbitListener(queues = "QD")
  23.     private void receiveD(Message message, Channel channel){
  24.         String msg=new String(message.getBody());
  25.         log.info("当前时间：{}，死信队列消息：{}",new Date().toString(),msg);
  26.     }
  27. }
4.9.1.6结果

请求地址：http://localhost:8080/ttl/sendMsg/123456789987456321

当前时间=Mon Nov 06 16:30:42 GMT+08:00 2023，发送一条信息给两个ttl队列：123456789987456321
当前时间：Mon Nov 06 16:30:52 GMT+08:00 2023，死信队列消息：消息来自ttl为10s的队列：123456789987456321
当前时间：Mon Nov 06 16:31:22 GMT+08:00 2023，死信队列消息：消息来自ttl为40s的队列：123456789987456321

可以看到 延时接收已经生效了，但是这个代码扩展性并不好，如果我们现在需要新增一个1小时以后得延时队列，还需要新建一个队列，重新建立链接等等，如果有无数个延时需求，就需要无数个队列来满足需求，所以扩展性并不好，也不现实。

4.9.2 延时队列优化

基于上述问题，我们应该创建一个通用的延迟队列，不设置过期时间，具体过期时间应该由生产者发消息时进行指定。这样就可以用一个延迟队列实现所有延迟需求。

4.9.2.1 新增config类QueueC

QueueC不设置过期时间，生命队列并绑定。

  1.  
  2. //声明称普通队列QC
  3.     @Bean("queueC")
  4.     public Queue queueC(){
  5.         Map<String, Object> arg=new HashMap<>(2);
  6.         //死信交换机
  7.         arg.put("x-dead-letter-exchange",Y_DEAD_EXCHANGE);
  8.         //死信RoutingKey
  9.         arg.put("x-dead-letter-routing-key","YD");
  10.         return QueueBuilder.durable(QUEUE_C).withArguments(arg).build();
  11.     }
  12.  
  13. //绑定queueC和xExchange
  14.     @Bean
  15.     public Binding queueCBindingX(@Qualifier("queueC") Queue queueC, @Qualifier("xExchange") DirectExchange
  16.             xExchange){
  17.         return BindingBuilder.bind(queueC).to(xExchange).with("XC");
  18.     }
4.9.2.2 创建生产者

创建生产者发送消息并设置过期时间

  1.  
  2. //开始发消息 以及TTL
  3.     @GetMapping("/sendExpiratMsg/{message}/{ttlTime}")
  4.     public void sendExpiratMsg(@PathVariable String message,@PathVariable String ttlTime){
  5.         log.info("当前时间={}，发送一条过期市场为：{}ms的信息给队列QC，信息：{}",new Date().toString(),ttlTime,message);
  6.         rabbitTemplate.convertAndSend("X","XC",message,msg->{
  7.             //设置发送消息的延迟时长
  8.             msg.getMessageProperties().setExpiration(ttlTime);
  9.             return msg;
  10.         } );
  11.     }

当前时间=Mon Nov 06 16:55:01 GMT+08:00 2023，发送一条过期市场为：200
当前时间：Mon Nov 06 16:55:04 GMT+08:00 2023，死信队列消息：你好2
当前时间=Mon Nov 06 16:55:08 GMT+08:00 2023，发送一条过期市场为：200
当前时间：Mon Nov 06 16:55:28 GMT+08:00 2023，死信队列消息：你好2

结果已经达到了动态设置过期时间。

4.9.2.3 注意

如果我们先发送20秒的消息你好1，然后发送2秒的消息你好2，他并不会先接收到你好2，因为消息队列只会检测第一条信息是否过期，并不会检测第二条信息是否过期，所以你好2会在你好1被接收后紧接着被接收。这是死信队列巨大的问题，因为你好1的时间不应该约束到你好2。

当前时间=Mon Nov 06 16:54:10 GMT+08:00 2023，发送一条过期市场为：20000
当前时间=Mon Nov 06 16:54:14 GMT+08:00 2023，发送一条过期市场为：2000m
当前时间：Mon Nov 06 16:54:30 GMT+08:00 2023，死信队列消息：你好1
当前时间：Mon Nov 06 16:54:30 GMT+08:00 2023，死信队列消息：你好2

4.9.3 RabbitMQ插件实现延迟队列

只要是基于死信队列的，上述问题都没办法处理，所以只能基于插件实现延迟队列。

4.9.3.1 下载插件

插件地址：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/

4.9.3.2 进入目录
cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.*.*/plugins
4.9.3.3 上传文件
4.9.3.4 安装插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
4.9.3.5 重启MQ
service rabbitmq-server restart
4.9.3.6 新建交换机

安装成功之后，新建交换机，发现type多了一个延迟消息类型。所以延迟消息不再由队列来实现，而是由交换机来实现。中间省去了死信队列的步骤。

4.9.3.7 新增配置类
  1. package com.rabbitmq02.config;
  2.  
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.CustomExchange;
  6. import org.springframework.amqp.core.Queue;
  7. import org.springframework.beans.factory.annotation.Qualifier;
  8. import org.springframework.context.annotation.Bean;
  9. import org.springframework.context.annotation.Configuration;
  10.  
  11. import java.util.HashMap;
  12. import java.util.Map;
  13.  
  14. /***
  15.  * @Description
  16.  * 给予插件的延迟队列
  17.  * @ClassName DelayedQueueConfig
  18.  * @Author LY
  19.  * @Date 2023/11/7 10:06
  20.  **/
  21. @Configuration
  22. public class DelayedQueueConfig {
  23.     
  24.     //交换机
  25.     public static final String DELAYED_EXCHANGE_NAME="delayed.exchange";
  26.     //队列
  27.     public static final String DELAYED_QUEUE_NAME="delayed.queue";
  28.     //ROUTINGKEY
  29.     public static final String DELAYED_ROUTING_KEY="delayed.routingkey";
  30.  
  31.     //声明队列
  32.     @Bean
  33.     public Queue delayedQueue(){
  34.         return new Queue(DELAYED_QUEUE_NAME);
  35.     }
  36.  
  37.     //声明交换机 基于插件
  38.     @Bean
  39.     public CustomExchange delayedExchange(){
  40.         /**
  41.          * 参数
  42.          * 1.交换机名称
  43.          * 2.交换机类型
  44.          * 3.是否需要持久化
  45.          * 4.是否需要自动删除
  46.          * 5.自定义参数
  47.          *
  48.          */
  49.         Map<String,Object> arg=new HashMap<>();
  50.         arg.put("x-delayed-type","direct");
  51.         return new CustomExchange(DELAYED_EXCHANGE_NAME,"x-delayed-message",true,false,arg);
  52.     }
  53.     //绑定
  54.     @Bean
  55.     public Binding delayedQueueBindingDelayedExchange(@Qualifier("delayedQueue") Queue delayedQueue,@Qualifier("delayedExchange") CustomExchange delayedExchange) {
  56.         return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
  57.     }
  58.  
  59. }
4.9.3.8 生产者
  1.  //基于插件的延时消息
  2.     @GetMapping("/sendDelayMsg/{message}/{delayTime}")
  3.     public void sendDelayMsg(@PathVariable String message,@PathVariable Integer delayTime){
  4.         log.info("当前时间={}，发送一条延时时间为：{}ms给队列delayed.queue，信息：{}",new Date().toString(),delayTime,message);
  5.         rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME,DelayedQueueConfig.DELAYED_ROUTING_KEY,message, msg->{
  6.             //设置发送消息的延迟时长
  7.             msg.getMessageProperties().setDelay(delayTime);
  8.             return msg;
  9.         } );
  10.     }
9.3.8 消费者
  1. @Component
  2. @Slf4j
  3. public class DelayQueueConsumer {
  4.     //接收消息
  5.     @RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)
  6.     private void receiveDelayedQueue(Message message ){
  7.         String msg=new String(message.getBody());
  8.         log.info("当前时间：{}，延迟队列消息：{}",new Date().toString(),msg);
  9.     }
  10. }
4.9.3.9 结论

基于插件的延迟消息可以做到完全根据发送消息延迟时间来进行延迟而不受消息顺序影响

当前时间=Tue Nov 07 10:53:45 GMT+08:00 2023，发送一条延时时间为：20000ms给队列delayed.queue，信息：你好1
当前时间=Tue Nov 07 10:53:49 GMT+08:00 2023，发送一条延时时间为：2000ms给队列delayed.queue，信息：你好2
当前时间：Tue Nov 07 10:53:51 GMT+08:00 2023，延迟队列消息：你好2
当前时间：Tue Nov 07 10:54:05 GMT+08:00 2023，延迟队列消息：你好1

4.9.4 总结

延时队列在需要延时处理的情况下非常有用，使用RabbitMQ来实现延迟队列可以很好的利用RabbitMQ的特性，如：消息可靠发送，消息可靠投递，死信队列来保障消息至少被消费一次，以及未被正确处理的消息不会被丢弃，另外通过RabbitMQ的集群特性，可以很好的解决单点故障问题，不会因为单节点挂掉导致延时队列不可用或者消息丢失。

当然延时队列还有其他很多西安则，例如JAVA的DelayQueue，利用Redis的zset利用Quartz或者kafka的时间轮，这些方式各有特点，具体要看使用的场景

5 发布确认高级

生产环境由于某些不明原因，导致rabbitMQ重启，在RabbitMQ重启期间生产者投递消息失败，导致消息丢失，需要手动处理和回复。于是我们考试思考，如何才能进行RabbitMQ的可靠投递。特别是在极端的情况下。RabbitMQ集群不可用的时候，无法投递的消息该如何处理呢。

5.1发布确认SpringBoot版

5.1.1确认机制方案

当交换机丢失，发送的消息自然就丢失了，交换机存在，队列丢失了，交换机又无法投递到队列，此时消息依然会被丢弃。

所以生产者将消息发送给交换机或者队列（MQ），无论交换机无法收到，亦或者无法投递给队列，生产者都应该将消息存入缓存中，然后采用定时任务对失败的消息重新发送。

5.1.2 代码架构图

这种架构会有两种问题：1：交换机出现问题。2：队列出现问题。

5.1.3 生产者

  1. package com.rabbitmq02.controller;
  2.  
  3. import com.rabbitmq02.config.ConfirmConfig;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.web.bind.annotation.GetMapping;
  8. import org.springframework.web.bind.annotation.PathVariable;
  9. import org.springframework.web.bind.annotation.RequestMapping;
  10. import org.springframework.web.bind.annotation.RestController;
  11.  
  12. @RequestMapping("/confirm")
  13. @RestController
  14. @Slf4j
  15. public class ProducerController {
  16.     @Autowired
  17.     private RabbitTemplate rabbitTemplate;
  18.     //发消息
  19.     @GetMapping("/sendMsg/{message}")
  20.     public void sendMsg(@PathVariable String message){
  21.            rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_ROUTING_KEY,message,correlationData);
  22.         rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_ROUTING_KEY+"11",message,correlationData);
  23.         rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME+"123",ConfirmConfig.CONFIRM_ROUTING_KEY,message,correlationData);
  24.         
  25.  
  26.         log.info("发送消息内容：{}",message);
  27.     }
  28. }

5.1.4 消费者

  1. package com.rabbitmq02.consumer;
  2.  
  3. import com.rabbitmq02.config.ConfirmConfig;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.amqp.core.Message;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.stereotype.Component;
  8.  
  9. /***
  10.  * @Description
  11.  * 发布确认高级 消费者
  12.  * @ClassName ConfirmConsumer
  13.  * @Author LY
  14.  * @Date 2023/11/7 11:37
  15.  **/
  16.  
  17. @Component
  18. @Slf4j
  19. public class ConfirmConsumer {
  20.     @RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE_NAME)
  21.     public void confirmMsg(Message message){
  22.        String msg= new String(message.getBody());
  23.         log.info("{}队列，接收到的消息内容：{}",ConfirmConfig.CONFIRM_QUEUE_NAME,msg);
  24.     }
  25. }

5.1.5 增加配置

在application.properties中添加

spring.rabbitmq.publisher-confirm-type=correlated

有三个可选模式

NONE：

禁用发布确认模式（默认）

CORRELATED：

发布消息成功到交换器后会触发回调方法

SIMPLE：

经测试有两种效果：

1.和CORRELATED一样，发布消息成功到交换器后会触发回调方法。

2.发布消息成功后使用RabbitTemplate调用waitFprConfirms或者waitForConfirmsOrDie方法等到broker节点返回发送结果，根据返回结果判定喜爱一步逻辑。要注意的是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker。相当于单个确认。

5.1.6 回调接口

当生产者消息发送成功以后，并不能感知到消息是否发送成功，所以应当提供一个会点接口，供消息投递成功以后来回调。

  1. package com.rabbitmq02.config;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.amqp.rabbit.connection.CorrelationData;
  5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.stereotype.Component;
  8.  
  9. import javax.annotation.PostConstruct;
  10.  
  11. @Component
  12. @Slf4j
  13. public class MyCallBack implements RabbitTemplate.ConfirmCallback {
  14.     //缺少入住回RabbitTemplate，如果不注入，那还是钓不到已经修改过的接口
  15.     @Autowired
  16.     private RabbitTemplate rabbitTemplate;
  17.     /**
  18.      * 注入
  19.      *  PostConstruct注解在其他注解完成后才会执行
  20.      */
  21.     @PostConstruct
  22.     public void init(){
  23.         rabbitTemplate.setConfirmCallback(this);
  24.     }
  25.     /**
  26.      * 交换机确认回调方法
  27.      * 1.发消息交换机接收成功
  28.      * @param correlationData  保存消息的id及相关的信息
  29.      * @param b 交换机是否收到消息 true
  30.      * @param s 原因  null
  31.      * 2.发消息交换机接收失败
  32.      * @param correlationData  保存消息的id及相关的信息
  33.      * @param b 交换机是否收到消息 false
  34.      * @param s 原因
  35.      */
  36.     @Override
  37.     public void confirm(CorrelationData correlationData, boolean b, String s) {
  38.             String id = correlationData==null?"":correlationData.getId();
  39.         if (b){
  40.             log.info("交换机已经收到了消息，id:{}",id);
  41.         }else {
  42.             log.info("交换机未收到消息，id:{}，原因：{}",id,s);
  43.         }
  44.     }
  45. }

 5.1.7 结果分析

正确的交换机，正确的routingKey，交换机收到了消息，队列接收到了消息，调用了正确的回调。

交换机：confirm.exchange，发送消息内容：你好2，routingKey:key1
confirm.queue队列，接收到的消息内容：你好2
正确的回调函数，交换机已经收到了消息，id:1

错误的交换机，正确的routingKey，交换机未收到消息，队列也为未收到消息，并有错误的回调。

交换机：confirm.exchange123，发送消息内容：你好2，routingKey:key1

Shutdown Signal: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'confirm.exchange123' in vhost '/', class-id=60, method-id=40)

错误的回调函数，交换机未收到消息，id:1，原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'confirm.exchange123' in vhost '/', class-id=60, method-id=40)

正确的交换机，错误的的routingKey，交换机接收到了消息，队列未收到消息，调用了正确的回调。

交换机：confirm.exchange，发送消息内容：你好2，routingKey:key111

正确的回调函数，交换机已经收到了消息，id:1

所以该方式只能确保消息正确到达了交换机，而不发保证是否真正的被队列所接收。

5.2回退消息

5.2.1 Mandatory参数

在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给生产者发送确认消息，如果发现该消息不可路由（无法发送到队列），那么该消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。通过设置Mandatory参数可以将消息在传递过程中不可到达目的地的消息返回给生产者。

5.2.2 新增配置

spring.rabbitmq.publisher-returns=true

5.2.3 新增配置

可以实现RabbitTemplate.ReturnCallback接口或者RabbitTemplate.ReturnsCallback接口。

  1. package com.rabbitmq02.config;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.amqp.core.Message;
  5. import org.springframework.amqp.core.ReturnedMessage;
  6. import org.springframework.amqp.rabbit.connection.CorrelationData;
  7. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.stereotype.Component;
  10.  
  11. import javax.annotation.PostConstruct;
  12.  
  13. @Component
  14. @Slf4j
  15. public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback  {
  16.     //缺少入住回RabbitTemplate，如果不注入，那还是钓不到已经修改过的接口
  17.     @Autowired
  18.     private RabbitTemplate rabbitTemplate;
  19.     /**
  20.      * 注入
  21.      *  PostConstruct注解在其他注解完成后才会执行
  22.      */
  23.     @PostConstruct
  24.     public void init(){
  25.         rabbitTemplate.setConfirmCallback(this);
  26.         rabbitTemplate.setReturnCallback(this);
  27.     }
  28.     /**
  29.      * 交换机确认回调方法
  30.      * 1.发消息交换机接收成功
  31.      * @param correlationData  保存消息的id及相关的信息
  32.      * @param b 交换机是否收到消息 true
  33.      * @param s 原因  null
  34.      * 2.发消息交换机接收失败
  35.      * @param correlationData  保存消息的id及相关的信息
  36.      * @param b 交换机是否收到消息 false
  37.      * @param s 原因
  38.      */
  39.     @Override
  40.     public void confirm(CorrelationData correlationData, boolean b, String s) {
  41.             String id = correlationData==null?"":correlationData.getId();
  42.         if (b){
  43.             log.info("正确的回调函数，交换机已经收到了消息，id:{}",id);
  44.         }else {
  45.             log.info("错误的回调函数，交换机未收到消息，id:{}，原因：{}",id,s);
  46.         }
  47.     }
  48.     /**
  49.      * 生产者发消息 如果消息没有被对应的交换机进队列
  50.      * 就会把消息回退给生产者 进行重发
  51.      * 在发送消息的过程中不可达目的地时将消息返回给生产者
  52.      */
  53.  
  54.  
  55.     @Override
  56.     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
  57.         log.info("消息{}，被交换机{}退回，退回的原因是{},路由key是{}",
  58.                 new String(message.getBody()),
  59.                 exchange, replyText, routingKey);
  60.  
  61.     }
  62.  
  63.  
  64.  
  65.    /* @Override
  66.     public void returnedMessage(ReturnedMessage returnedMessage) {
  67.         log.info("消息{}，被交换机{}退回，退回的原因是{},路由key是{}",
  68.                 new String(returnedMessage.getMessage().getBody()),
  69.                 returnedMessage.getExchange(), returnedMessage.getReplyText(), returnedMessage.getRoutingKey());
  70.     }
  71. */
  72.  
  73. }

5.2.4 结果分析

正确的交换机，正确的routingKey，交换机接收成功，队列接收成功，消费者接收成功，正确的回调。

交换机：confirm.exchange，发送消息内容：你好2，routingKey:key1

confirm.queue队列，接收到的消息内容：你好2
正确的回调函数，交换机已经收到了消息，id:1

正确的交换机，错误的routingKey，交换机接受成功，队列未接收，消费者未接收，正确的回调函数，消息被退回给生产者。

交换机：confirm.exchange，发送消息内容：你好2，routingKey:key111
消息你好2，被交换机confirm.exchange退回，退回的原因是NO_ROUTE,路由key是key111
正确的回调函数，交换机已经收到了消息，id:1

5.3 备份交换机

有了mandatory参数和回退消息，我们获得了对无法投递消息的感知能力，有机会再生产这的消息无法被投递时发现并处理。但是有时候，我们并不知道无法处理这些消息，最多打印一下日志，然后触发警报，手动处理。而且无形之中增加了生产者的复杂性，如果生产者在多个服务器的时候，手动复制日志也容易出错。

而几不想增加生产者的复杂性，又想处理这些无法被路由的消息，我们可以设置一个交换机为另一个交换机的备份交换机，当见换机收到一条无法路由的消息时，把该条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机类型为Fanout，这样就能把所有的消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样原本交换机所有无法路由的消息都会进入到该队列中，也可以在新建一个报警队列，用独立的消费者来进行监测和报警

5.3.1 代码架构图

 5.3.2 修改配置文件

在高级发布确认模式的基础上，新增备份交换机，备份队列，报警队列

  1. package com.rabbitmq02.config;
  2.  
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.beans.factory.annotation.Qualifier;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  7.  
  8. import java.util.HashMap;
  9.  
  10. /***
  11.  * @Description
  12.  * 发布确认 高级 配置类
  13.  * @ClassName ConfirmConfig
  14.  * @Author LY
  15.  * @Date 2023/11/7 11:25
  16.  **/
  17. @Configuration
  18. public class ConfirmConfig {
  19.     //交换机
  20.     public static final String CONFIRM_EXCHANGE_NAME="confirm.exchange";
  21.     //队列
  22.     public static final String CONFIRM_QUEUE_NAME="confirm.queue";
  23.     //ROUTINGKEY
  24.     public static final String CONFIRM_ROUTING_KEY="key1";
  25.     //备份交换机
  26.     public static final String BACKUP_EXCHANGE_NAME="backup.exchange";
  27.     //备份队列
  28.     public static final String BACKUP_QUEUE_NAME="backup.queue";
  29.     //报警队列
  30.     public static final String WARNING_QUEUE_NAME="warning.queue";
  31.  
  32.     //创建备份交换机
  33.     @Bean
  34.     public FanoutExchange buckupExchange(){
  35.         return new FanoutExchange(BACKUP_EXCHANGE_NAME);
  36.     }
  37.     //声明交换机
  38.     @Bean
  39.     public DirectExchange confirmExchange(){
  40.         //return new DirectExchange(CONFIRM_EXCHANGE_NAME);
  41.         return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true).withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
  42.     }
  43.  
  44.     //声明队列
  45.     @Bean
  46.     public Queue confirmQueue(){
  47.         return new Queue(CONFIRM_QUEUE_NAME);
  48.     }
  49.     //创建备份队列
  50.     @Bean
  51.     public Queue backupQueue(){
  52.         return new Queue(BACKUP_QUEUE_NAME);
  53.     }
  54.     //创建报警队列
  55.     @Bean
  56.     public Queue warningQueue(){
  57.         return new Queue(WARNING_QUEUE_NAME);
  58.     }
  59.     //绑定
  60.     @Bean
  61.     public Binding bindconfirmExchangeToConfirmQueue(@Qualifier("confirmExchange") DirectExchange confirmExchange,@Qualifier("confirmQueue") Queue confirmQueue){
  62.         return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
  63.     }
  64.  
  65.  
  66.     //绑定备份交换机和备份队列
  67.     @Bean
  68.     public Binding buckupExchangeBindToBackupQueue(@Qualifier("buckupExchange") FanoutExchange buckupExchange,@Qualifier("backupQueue") Queue backupQueue){
  69.         return BindingBuilder.bind(backupQueue).to(buckupExchange);
  70.     }
  71.     //绑定备份交换机和报警队列
  72.     @Bean
  73.     public Binding buckupExchangeBindToWarningQueue(@Qualifier("buckupExchange") FanoutExchange buckupExchange,@Qualifier("warningQueue") Queue warningQueue){
  74.         return BindingBuilder.bind(warningQueue).to(buckupExchange);
  75.     }
  76. }

5.3.3 新增报警消费者

  1. package com.rabbitmq02.consumer;
  2.  
  3. import com.rabbitmq02.config.ConfirmConfig;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.amqp.core.Message;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.stereotype.Component;
  8.  
  9. /***
  10.  * @Description
  11.  * 发布确认高级 消费者
  12.  * @ClassName ConfirmConsumer
  13.  * @Author LY
  14.  * @Date 2023/11/7 11:37
  15.  **/
  16.  
  17. @Component
  18. @Slf4j
  19. public class WarningConsumer {
  20.     @RabbitListener(queues = ConfirmConfig.WARNING_QUEUE_NAME)
  21.     public void confirmMsg(Message message){
  22.        String msg= new String(message.getBody());
  23.         log.info("{}报警队列队列，接收到的消息内容：{}",ConfirmConfig.WARNING_QUEUE_NAME,msg);
  24.     }
  25. }

5.3.4 结果分析

正确的交换机，正确的routingKey还是正常接收。

交换机：confirm.exchange，发送消息内容：你好2，routingKey:key1
正确的回调函数，交换机已经收到了消息，id:1
confirm.queue队列，接收到的消息内容：你好2
 

正确的交换机，错误的routingKey，则通过备份交换机，被发到了报警队列中。

交换机：confirm.exchange，发送消息内容：你好2，routingKey:key111
warning.queue报警队列队列，接收到的消息内容：你好2

mandatory参数与备份交换机一起使用的时候，备份交换机优先级较高。两者同时存在不会被退回到生产者。

6.RabbitMQ其他知识点

6.1幂等性

6.1.1概念

用户对于同一操作发起的一次货多次请求的结果是一致的，不会因为多次点击而产生副作用。例如支付，用户购买商品后支付，支付扣款成功，但是返回结果的时网络异常，此时钱已经扣了，用户再次点击，此时会第二次扣款，返回结果成功，此时用户被扣款两次，流水记录也是两次，在以前的系统中，我们只需要把数据操作放入事物中即可，发生错误立即回滚，但是响应客户端的时候也可能出现某些问题。

6.1.2 消息重复消费

消费者在消费MQ中的消息时，MQ已经把消息发送给消费者，消费者再给MQ返回ack时网络中断，故MQ未收到确认消息，该消息会被转发给其他消费者，或网络重连后重新发给该消费者，但实际上该消费者已经成功消费了该消息，造成了消费者消费了重复的消息。

6.1.3 解决思路

MQ消费者的幂等性问题解决一般使用全局ID，或者写一个唯一标识等等，每次消费时都先通过该id或标识判断是否已经消费过。

6.1.4 消费端的幂等性保障

在业务高分时期，生产端可能重复发送了消息，这时候消费端就要实现幂等性，这就意味着我们的消息永远不会被消费多次，业内主流的幂等性操作有两种：A.唯一ID+指纹码机制。B.利用redis的原子性去实现。

6.1.5 唯一ID+指纹码

指纹码：我们的一些规则或者时间戳加别的服务给到的唯一信息码，但他不一定是我们系统生成的，基本都是有我们业务规则拼接而来，但必须要保证唯一性，然后利用查询语句进行判断这个id是否已存在，优势是实现简单，就一个拼接，然后查询判重。劣势是在高并发时如果是单个数据库就会有写入性瓶颈，也可以用分库分表提升性能，但是并不推荐。

6.1.6 Redis原子性

利用redis执行setnx命令，天然具有幂等性判断，从而实现不重复消费。

6.2 优先级队列

6.2.1 使用场景

普通商城系统中，有一个订单催促功能，例如客户下单后，特定时间内未付款，就会发送短信提醒，但是商家肯定要区分一些大商家跟小商家，大商家给带来的利润相对较大，所以他们的订单理所应当的优先处理，之前都是使用Redis的List做一个简单的消息队列，并不能实现优先级的场景。

所以当订单量大了以后，采用RabbitMQ进行改造和优化，如果是大客户推给一下相对比较高的有衔接，否则就是默认优先级。

6.2.2 如何添加

6.2.2.1 页面添加

优先级取值可以是0-255 太大会造成性能下降。

6.2.2.2 代码添加

生产者：

给5号设置为最高优先级

  1. package com.rabbitmq.priorityQueue;
  2.  
  3. import com.rabbitmq.client.AMQP;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.utils.RabbitMQUtil;
  6.  
  7. import java.io.IOException;
  8. import java.util.HashMap;
  9. import java.util.Map;
  10. import java.util.concurrent.TimeoutException;
  11.  
  12. public class Producer {
  13.     public static final String QUEUE_NAME="hello";
  14.     public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
  15.         Channel channel = RabbitMQUtil.getChannel();
  16.         Map<String,Object> arg=new HashMap<>();
  17.         arg.put("x-max-priority",10);
  18.         channel.queueDeclare(QUEUE_NAME,false,false,false,arg);
  19.         for (int i = 1; i <= 20; i++) {
  20.             String massage="hello word"+i;
  21.             if (i==5){
  22.                 AMQP.BasicProperties properties=new AMQP.BasicProperties().builder().priority(5).build();
  23.  
  24.                 channel.basicPublish("",QUEUE_NAME,properties, massage.getBytes());
  25.             }else {
  26.             channel.basicPublish("",QUEUE_NAME,null, massage.getBytes());
  27.             }
  28.             System.out.println("消息已发送完毕"+massage);
  29.         }
  30.     }
  31. }

消费者：

  1. package com.rabbitmq.priorityQueue;
  2.  
  3. import com.rabbitmq.client.BuiltinExchangeType;
  4. import com.rabbitmq.client.CancelCallback;
  5. import com.rabbitmq.client.Channel;
  6. import com.rabbitmq.client.DeliverCallback;
  7. import com.rabbitmq.utils.RabbitMQUtil;
  8.  
  9. import java.io.IOException;
  10. import java.util.HashMap;
  11. import java.util.Map;
  12. import java.util.concurrent.TimeoutException;
  13.  
  14. public class Consumer01 {
  15.     public static final String QUEUE_NAME="hello";
  16.  
  17.     public static void main(String[] args) throws IOException, TimeoutException {
  18.         Channel channel = RabbitMQUtil.getChannel();
  19.         DeliverCallback deliverCallback=(consumerTag,massage)->{
  20.             System.out.println(new String(massage.getBody()));
  21.         };
  22.         //声明参数4 取消消息接收时的回调
  23.         CancelCallback cancelCallback= consumerTag->{
  24.             System.out.println("消息接收被中断");
  25.         };
  26.         channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
  27.     }
  28. }
6.2.2.3 结论 

5被第一个取出

hello word5
hello word1
hello word2
hello word3

6.3 惰性队列

正常情况下消息是保存在内存中的，但是队形队列的消息是保存在磁盘中的。效率并不高。

6.3.1 使用场景

RabbitMQ从3.6.0版本开始引入了队形队列的概念，惰性队列会尽可能的将消息存入磁盘中，而在消费者消费到相应的消息时才会被加载到内存中，他的一个重要设计目标是支持更长的队列，既更多的消息存储，当消费者由于各种各样的原因（宕机，关闭等）导致长时间不能消费消息，造成消息堆积时，惰性队列就很有必要了。

默认情况下，生产者将消息发送到MQ时，队列中的消息会尽可能存到内存中，这样可以更快速的将消息发送给消费者，即使是持久化的信息，再被写入磁盘中同时也会在内存中驻留一份备份。当MQ释放内存的时候才会将内存中的消息写入磁盘中，这个操作消耗时间过长，也会阻塞队列操作，进而无法接受新的消息。MQ的开发者们一直在升级相关算法，但是效果始终不太理想，尤其是在信息量特别大的时候。

6.3.2 两种模式

队列具有两种模式：default和lazy，默认的为default模式，在3.6.0之前无需做任何变更。lazy模式即为惰性队列模式，可以通过调用channel.queueDeclare方法的时候在参数中设置，也可以通过Policy的方式设置，如果一个队列同时使用两种方式设置的话，Policy的方式有更高的优先级，如果要通过声明的方式改变已有队列的话，只能先删除已有队列，再重新声明。

5.3.2.1 客户端设置

5.3.2.2 代码设置

队列生命的时候可以通过“x-queue-mode”参数来设置队列模式，取值为“default”和“lazy”：

  1.          Map<String,Object> arg=new HashMap<>();
  2.         arg.put("x-queue-mode","lazy");
  3.         channel.queueDeclare(QUEUE_NAME,false,false,false,arg);

6.3.3 内存开销对比

在发送100w条数据的情况下，普通队列消耗内存1.2g，惰性队列仅仅占用1.2m。

7 RabbitMQ 集群

7.1 使用集群的原因

RabbitMQ遇到内存崩溃，机器掉电或者主板故障等问题或者单台MQ服务器可以满足1000条消息吞吐量，但是需要10w条消息吞吐量，够慢昂贵的服务器来增强MQ性能并不可取，所以可以选择搭建集群来解决实际问题。

7.1.2 搭建集群

可参考其他文章

7.2 镜像队列

如果RabbitMQ集群中只有一个Broker节点，那么该节点失效将导致整体服务临时性不可用，并且可能导致消息丢失，就算将消息设置为持久化，对应队列也持久化，仍然无法避免消息发送后和写入磁盘之前出现问题。

引入镜像队列（Mirror Queue）机制，可以将队列镜像到集群中的其他Broker节点上，如果集群中的一个节点失效了，就可以自动切换到其他节点上，保证服务的可用性。

7.2.1 搭建步骤 

可参考其他文章

7.3 Haproxy+Keepalive 高负载高可用

当有一个队列无法连接，他的镜像队列可以连接，但是生产者只会连接到该无法连接该镜像队列，此时MQ无法处理，但是可以使用Haproxy+Keepalive 实现高可用。

HAProxy提供高可用性，负载均衡以及基于TCP/HTTP应用的代理，支持虚拟主机，他是免费，快速并可靠的一种解决方案，包括Twiter，Reddit，StackOverflow，GitHub在内的多家知名公司正在使用，Haproxy实现了一种事件驱动，单一进程模型，此模型支持非常大的并发连接数。

7.3.1 搭建步骤 

可参考其他文章

7.3.2 扩展

nginx，lvs，Haproxy，之间的区别：

www.ha97.com/5646.html

7.4 Federation Exchange

联邦交换机，联合交换机

7.4.1 概念

当我们有两个Broker，彼此之间相距较远，离Broker1近的客户端，应当访问Broker1，但是Broker1和Broker2之间也应当进行数据同步，否则离Broker1近的客户端无法获取到Broker2内的数据。

7.4.2 原理

7.5 Federation Queue

7.5.1 概念

联邦队列可以再多个Broker节点或者集群之前为单个队列提供负载均衡功能，一个联邦队列可以连接一个或多个上游队列，并从这些上游队列中获取消息，以满足本地消费者消费消息的需求。

7.5.2 原理

7.6 shovel

7.6.1 概念

Federation具备类似数据转发的功能，shovel能够可靠，持续的从一个Borker中的队列（作为源端：source）拉去数据并转发至另一个Borker中的交换器（作为目的端：destination）。作为源端的队列和作为目的端的交换器可以同时位于同一个Broker也可以在不同的Broker上。Shovel可以翻译为“铲子”，是一种比较形象的比喻，铲子可以将消息从一方铲向另一方。shovel行为就像优秀的客户端程序，能够负责连接源和目的地，负责消息的读写及负责连接失败问题的处理。

7.6.2 原理

7.6.1 搭建步骤 

可参考其他文章

完.

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/从前慢现在也慢/article/detail/939026
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号