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

RabbitMQ_rabbitmq启动修改端口

作者：IT小白 | 2024-05-12 08:56:01

赞

踩

rabbitmq启动修改端口

测试代码——消费者1

一、rabbitmq的安装及启动

1.下载对应版本的erlang和rabbitmq文件。(第二推荐)

或者你可以打开下面的网站找到你想要的版本（推荐，使用此方法可跳过2、3两步）

在创建文件夹并进入后输入页面的命令，例如：

  1. #下载erlang
  2. curl -s https://packagecloud.io/install/repositories/rabbitmq/erlang/script.rpm.sh | sudo bash
  3. #安装erlang
  4. sudo yum install erlang-debuginfo-23.3.4.11-1.el7.x86_64
  5. #下载rabbitmq
  6. curl -s https://packagecloud.io/install/repositories/rabbitmq/rabbitmq-server/script.rpm.sh | sudo bash
  7. #安装rabbitmq
  8. sudo yum install rabbitmq-server-3.10.0-1.el7.noarch

下载地址：rabbitmq/erlang - Packages · packagecloud

rabbitmq/rabbitmq-server - Packages · packagecloud

查看对应版本：RabbitMQ Erlang Version Requirements — RabbitMQ

我这里使用的本虚拟机是centos7于是选择了这两个版

2.在虚拟机中使用mkdir命令创建文件夹 ——/root/rabbitmq（使用推荐方法后跳过此步骤）

3.把文件传输到虚拟机中（使用推荐方法后跳过此步骤）

首先使用ip addr命令查看虚拟机的ip

 在物理机上win+r -->cmd -->scp "C:/Users/q/Downloads/rabbitmq-server-3.10.0.rc.3-1.el7.noarch.rpm" root@172.20.10.2:/root/rabbitmq

 在执行命令前物理机会尝试连接虚拟机，如果超时可以参考如下三个教程（练习环境中可以关闭防火墙，如是生产环境，为了网络安全，请看其他教程开放15672及5672端口）

Centos怎么关闭防火墙？Centos7关闭防火墙详细教程 -Win7系统之家

CentOS7配置IP_centos7配置ip地址_凡梦千华的博客-CSDN博客

Permission denied (publickey,gssapi-keyex,gssapi-with-mic)_H.S.T不想卷的博客-CSDN博客

在执行命令时会要求你输入密码，输入对应用户的密码就可以了。

4.安装完成启动rabbitmq

  1. #安装rabbitmq
  2. yum localinstall rabbitmq-server3.8.6-1.el7.noarch.rpm

 注意e后面是小的l不是1，当出现如下页面时输入y并按回车键

输入如下命令启动并查看rabbitmq服务的状态

  1. #启动rabbitmq
  2. systemctl start rabbitmq-server
  3. #查看rabbitmq的状态
  4. systemctl status rabbitmq-server

 显示如下内容则为rabbitmq已经成功启动

5.安装rabbitmq管理页面及创建用户

 默认情况下rabbimq没有安装web端的客户端软件，需要安装才能生效

  1. # 打开rabbitmq管理界面插件
  2. rabbitmq-plugins enable rabbitmq_management

安装完成后即可通过服务器ip:15672打开管理后台

如果此时你使用的是服务器本机那么可以使用默认用户guest访问，如果使用其他电脑那么需要新创建一个用户并授权

 6.新建远程用户

操作权限分为下列四种：
administrator：（管理员）可以登录控制台、查看所有信息、并对rabbitmq进行管理
monToring：（监控者）可以登录控制台，查看所有信息
policymaker：（策略制定者）可以登录控制台指定策略
managment：（普通管理员）可以进行登录控制

  1. #添加用户
  2. rabbitmqctl add_user 用户名 密码
  3.  
  4. #设置用户角色，并分配操作权限
  5. rabbitmqctl set_user_tags 用户名 角色
  6.  
  7. #为用户添加资源权限（授予虚拟机根节点的所有权限）
  8. rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*"

7.修改默认端口

使用whereis reabbitmq查看rabbitmq的安装路径

新增配置文件（在/etc/rabbitmq目录下）

vim /etc/rabbitmq/rabbitmq.conf

进入文本编辑器后按i键输入如下配置即更改了通讯端口为5673 

输入如下命令修改默认配置,请根据你们的rabbitmq版本对命令进行修改

输入结束后按ESC键，输入:wq并按回车键

重启rabbitmq即完成修改，如想测试请更改management.tcp.port后的端口，重启后使用服务器ip+新端口号才可访问rabbitmq的管理后台

systemctl restart rabbitmq.server

以下步骤因本人测试后出现问题，无法启动rabbitmq，故不推荐修改：

vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.6/sbin/rabbitmq-defaults

 进入文本编辑器后按i键，将光标定位到BOOT_MODULE="rabbit"下面并输入(根据你们放置的路径修改)：

CONFIG_FILE=/etc/rabbitmq/rabbitmq.conf

 输入结束后按ESC键，输入:wq并按回车键

重启rabbitmq

systemctl restart rabbitmq-server

二、编写测试程序

1.新建一个maven工程

2.引入依赖（当我未引入slf4j-nop时程序启动报错SLF4J: Failed to load class “org.slf4j.impl.StaticLoggerBinder，引入改依赖后错误消失）

  1.     <dependencies>
  2.         <dependency>
  3.             <groupId>org.slf4j</groupId>
  4.             <artifactId>slf4j-nop</artifactId>
  5.             <version>1.7.2</version>
  6.  
  7.         </dependency>
  8.         <!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
  9.         <dependency>
  10.             <groupId>com.rabbitmq</groupId>
  11.             <artifactId>amqp-client</artifactId>
  12.             <version>5.18.0</version>
  13.         </dependency>
  14.  
  15.         <dependency>
  16.             <groupId>commons-io</groupId>
  17.             <artifactId>commons-io</artifactId>
  18.             <version>2.6</version>
  19.         </dependency>
  20.     </dependencies>

参考：Maven项目报错:“ SLF4J: Failed to load class “org.slf4j.impl.StaticLoggerBinder”的解决方案_java_脚本之家

3.编写 生产者代码（其中的异常既可以像下面这样写，也可以throws）

  1. public class Producer {
  2.  
  3.     public static final String QUEUE_NAME="hello";
  4.  
  5.     public static void main(String[] args) {
  6.         //创建一个连接工厂
  7.         ConnectionFactory factory=new ConnectionFactory();
  8.         //工厂IP连接RabbitMQ的队列
  9.         factory.setHost("172.20.10.2");
  10.         //用户名
  11.         factory.setUsername("admin");
  12.         //密码
  13.         factory.setPassword("root");
  14.  
  15.         try {
  16.             //创建连接
  17.             Connection connection = factory.newConnection();
  18.             //获取信道
  19.             Channel channel = connection.createChannel();
  20.  
  21.             /**
  22.              * 生成一个队列
  23.              * 1.队列名称
  24.              * 2.队列里面的消息是否持久化（写入磁盘）默认情况下是不持久化的（写入内存）
  25.              * 3.该队列是否只供一个消费者进行消费，是否进行消息共享，true可以多个消费者消费，false不能多个消费者消费
  26.              * 4.是否自动删除 最后一个消费者断开连接以后，该队列是否自动删除 true自动删除，false不自动删除
  27.              * 5.其他参数
  28.              */
  29.             AMQP.Queue.DeclareOk declareOk = channel.queueDeclare(QUEUE_NAME,false,false,false,null);
  30.             //发消息
  31.             String message="hello world";//初次使用
  32.  
  33.             /**
  34.              * 发送一个消费
  35.              * 1.发送到哪个交换机
  36.              * 2.路由的key值是哪个 本次队列的名称
  37.              * 3.其他参数消息
  38.              * 4.发送消息的消息体
  39.              */
  40.             channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
  41.  
  42.             System.out.println("发送完成~");
  43.         } catch (IOException e) {
  44.             throw new RuntimeException(e);
  45.         } catch (TimeoutException e) {
  46.             throw new RuntimeException(e);
  47.         }
  48.     }
  49. }

 编写完成后即开始测试，发送后rabbitmq后台可看到“hello”队列

4.编写消费者代码

  1. public class Consumer {
  2.  
  3.     //队列的名称
  4.     public static final String QUEUE_NAME="hello";
  5.  
  6.     //接收消息
  7.     public static void main(String[] args) {
  8.         //创建连接工厂
  9.         ConnectionFactory factory = new ConnectionFactory();
  10.         try {
  11.             factory.setHost("172.20.10.2");
  12.             factory.setUsername("admin");
  13.             factory.setPassword("root");
  14.  
  15.             //创建连接
  16.             Connection connection = factory.newConnection();
  17.             //创建信道
  18.             Channel channel = connection.createChannel();
  19.  
  20.             //声明接收消息
  21.             DeliverCallback deliverCallback = (consumerTag,message) ->{
  22.                 System.out.println(message.getBody().toString());
  23.             };
  24.  
  25.             //声明取消消息
  26.             CancelCallback cancel =(String var1) ->{
  27.                 System.out.println("接收失败");
  28.             };
  29.             /**
  30.              * 消费者消费消息
  31.              * 1.消费哪个队列
  32.              * 2.消费成功之后是否要自动应答 true代表的是自动应答，false代表手动应答
  33.              * 3.消费者未成功消费的回调
  34.              * 4.消费者取消费的回调
  35.              */
  36.             channel.basicConsume(QUEUE_NAME,true,deliverCallback, cancel);
  37.  
  38.         } catch (IOException e) {
  39.             throw new RuntimeException(e);
  40.         } catch (TimeoutException e) {
  41.             throw new RuntimeException(e);
  42.         }
  43.     }
  44. }

测试打印出来的是一个对象，修改输出语句为，输出即为可读的hello word

System.out.println(new String(message.getBody()));

 小插曲：win10设置打不开

1.按下WIN+X打开超级菜单，点击命令提示符（powershell管理员）

2.输入 sfc /SCANNOW 按下回车开始扫描

3.完成后重启即可解决

三、工作队列

1.编写测试生产者，使用控制台输入发送内容

  1. public class WorkQueueTest {
  2.  
  3.     public static void main(String[] args) throws Exception {
  4.         Channel channel = RabbitMQUtil.getChannel();
  5.  
  6.         channel.queueDeclare("hello", false, false, false, null);
  7.         Scanner scanner = new Scanner(System.in);
  8.  
  9.         while (scanner.hasNext()) {
  10.             String message = scanner.next();
  11.             channel.basicPublish("", "hello",null, message.getBytes());
  12.             System.out.println("发送完成");
  13.         }
  14.  
  15.     }
  16. }

2.启动两个消费者

启动一个消费者后点击如下按钮-->Edit Configurations-->Modify options-->勾选Allow multiple instances-->启动

 

 随后在控制台键入输入4次数据，并成功接收，测试结束。（轮训分发，公平分发）

四、消息应答

为了保证消息在发送过程中不丢失，rabbitmq引入消息应答。消息应答：消费者在接收到消息并处理该消息后，告诉rabbitmq它已经处理了，rabbitmq可以把该消息删除了。

当关闭消息自动应答时，消费者未ack，消息将自动重新入队。

推荐：关闭消息自动应答，关闭消息批量应答

1.测试

生产者：

  1. /**
  2.  * @author tom
  3.  * @create 2023-08-23 16:42
  4.  *
  5.  * 消息在手动应答时是不丢失、放回队列中重新消费
  6.  */
  7. public class Task2 {
  8.  
  9.     //信道名
  10.     public static final String QUEUE_NAME="ack_queue";
  11.  
  12.     public static void main(String[] args) throws Exception {
  13.  
  14.         //获取信道
  15.         Channel channel = RabbitMQUtil.getChannel();
  16.  
  17.         //声明队列
  18.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  19.  
  20.         //从控制台中输入信息
  21.         Scanner scanner = new Scanner(System.in);
  22.         while (scanner.hasNext()){
  23.             String message = scanner.next();
  24.             channel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));
  25.             System.out.println("生产者发出消息："+message);
  26.         }
  27.  
  28.  
  29.     }
  30. }

消费者1：

  1. /**
  2.  * @author tom
  3.  * @create 2023-08-23 17:06
  4.  * <p>
  5.  * 消息收到应答时不丢失、放回队列中重新消费
  6.  */
  7. public class Work02 {
  8.  
  9.     public static final String TASK_NAME = "ack_queue";
  10.  
  11.     public static void main(String[] args) throws Exception {
  12.  
  13.         Channel channel = RabbitMQUtil.getChannel();
  14.  
  15.         //消费成功回调
  16.         DeliverCallback deliver = (consumerTag, message) -> {
  17.             //沉睡1S
  18.             SleepUtils.sleep(1);
  19.             System.out.println("消息消费成功"+new String(message.getBody(),"UTF-8"));
  20.  
  21.             //手动应答
  22.             channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
  23.         };
  24.  
  25.         //消费失败回调
  26.         CancelCallback handle = (String var1) -> {
  27.             System.out.println("消息消费失败");
  28.         };
  29.  
  30.         boolean autoAck = false;
  31.         //消费信息
  32.         channel.basicConsume(TASK_NAME, autoAck, deliver, handle);
  33.  
  34.  
  35.     }
  36. }

消费者2：

  1. /**
  2.  * @author tom
  3.  * @create 2023-08-23 17:06
  4.  * <p>
  5.  * 消息收到应答时不丢失、放回队列中重新消费
  6.  */
  7. public class Work03 {
  8.  
  9.     public static final String TASK_NAME = "ack_queue";
  10.  
  11.     public static void main(String[] args) throws Exception {
  12.  
  13.         Channel channel = RabbitMQUtil.getChannel();
  14.  
  15.         //消费成功回调
  16.         DeliverCallback deliver = (consumerTag, message) -> {
  17.             //沉睡1S
  18.             SleepUtils.sleep(10);
  19.             System.out.println("消息消费成功"+new String(message.getBody(),"UTF-8"));
  20.  
  21.             //手动应答
  22.             channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
  23.         };
  24.  
  25.         //消费失败回调
  26.         CancelCallback handle = (String var1) -> {
  27.             System.out.println("消息消费失败");
  28.         };
  29.  
  30.         boolean autoAck = false;
  31.         //消费信息
  32.         channel.basicConsume(TASK_NAME, autoAck, deliver, handle);
  33.  
  34.  
  35.     }
  36. }

测试结果：

当发送44时关闭work03，消息没有被丢弃而是重新回到队列中

2. 开启队列持久化（在生产者处设置）

首先删除队列

点击对应队列 

 已开启持久化

 当队列被持久化后，即便重启rabbitmq，队列也依然存在

3.消息实现持久化（在生产者处设置）

  1. //设置生产者发送消息为持久化消息(保存到磁盘上)，当前无法完全确保消息持久化，需要完全支持化请搜索发布确认相关内容
  2.             channel.basicPublish("",QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));

4.不公平分发（在2个消费者处都设置一下）

  1.         //设置为不公平分发
  2.         int prefetchCount=1;
  3.         channel.basicQos(prefetchCount);

测试结果：

 

 

 因work02沉睡1s而work03沉睡10s，所以更多的消息被发送给了work02

5.预取值prefetch

设置消费者一次取几条消息

 设置预取值（同不公平分发的设置一样）

  1.         //预期值
  2.         int prefetchCount=5;
  3.         channel.basicQos(prefetchCount);

在信道推送消息时将按顺序推送，即轮流给两个消费者推送，当消费者未消费的信息等于预取值时，队列将直接将该消息推送给其他消费者

五、发布确认

发布确认的流程

 开启发布确认（生产者）

  1.         //开启发布确认
  2.         channel.confirmSelect();

发布确认 （生产者）

channel.waitForConfirms();
  1. package com.atguigu.rabbitmq.four;
  2.  
  3. import com.atguigu.rabbitmq.util.RabbitMQUtil;
  4. import com.rabbitmq.client.Channel;
  5. import com.rabbitmq.client.ConfirmCallback;
  6.  
  7. import java.util.UUID;
  8. import java.util.concurrent.ConcurrentHashMap;
  9.  
  10. /**
  11.  * @author tom
  12.  * @create 2023-08-31 9:04
  13.  */
  14. public class AsyncWorkQueue {
  15.  
  16.     static int MAX = 1000;
  17.  
  18.     public static void main(String[] args) throws Exception {
  19.         //单个发布确认
  20. //        publishMessageIndividually();     //600ms
  21.  
  22.         //批量发布确认
  23. //        BatchReleaseConfirmation();       //80ms
  24.  
  25.         //异步发布确认
  26.         AsyncReleaseConfirmation();         //35ms
  27.     }
  28.  
  29.     public static void publishMessageIndividually() throws Exception {
  30.  
  31.         //获取信道
  32.         Channel channel = RabbitMQUtil.getChannel();
  33.         //开启发布确认
  34.         channel.confirmSelect();
  35.  
  36.         //随机生成队列名
  37.         String queueName = UUID.randomUUID().toString();
  38.         //队列构造 队列名/是否持久化/是否供多个消费者消费/是否自动删除/其他参数
  39.         channel.queueDeclare(queueName, true, false, false, null);
  40.  
  41.         long begin = System.currentTimeMillis();
  42.  
  43.         int counter = 0;
  44.         for (int i = 0; i < MAX; i++) {
  45.             String message = "消息" + i;
  46.  
  47.             //发送到哪个交换机/队列名/其他参数/发送消息的消息体
  48.             channel.basicPublish("", queueName, null, message.getBytes("UTF-8"));
  49.  
  50.             //单个消息就马上进行发布确认
  51.             boolean flag = channel.waitForConfirms();
  52.             if (flag = true)
  53.                 counter++;
  54.         }
  55.  
  56.         if (counter == MAX) {
  57.             long end = System.currentTimeMillis();
  58.             long result = end - begin;
  59.             System.out.println("发送1000条消息使用了" + result + "ms");
  60.         }
  61.     }
  62.  
  63.     public static void BatchReleaseConfirmation() throws Exception {
  64.         //获取信道
  65.         Channel channel = RabbitMQUtil.getChannel();
  66.         //开启发布确认
  67.         channel.confirmSelect();
  68.  
  69.         //随机生成队列名
  70.         String queueName = UUID.randomUUID().toString();
  71.         //队列构造 队列名/是否持久化/是否供多个消费者消费/是否自动删除/其他参数
  72.         channel.queueDeclare(queueName, true, false, false, null);
  73.  
  74.         long begin = System.currentTimeMillis();
  75.  
  76.         int counter = 0;
  77.         for (int i = 0; i < MAX; i++) {
  78.             String message = "消息" + i;
  79.  
  80.             //发送到哪个交换机/队列名/其他参数/发送消息的消息体
  81.             channel.basicPublish("", queueName, null, message.getBytes("UTF-8"));
  82.  
  83.             if (i / 100 == 0) {
  84.                 channel.waitForConfirms();
  85.             }
  86.             counter++;
  87.         }
  88.  
  89.         if (counter == MAX) {
  90.             long end = System.currentTimeMillis();
  91.             long result = end - begin;
  92.             System.out.println("发送1000条消息使用了" + result + "ms");
  93.         }
  94.     }
  95.  
  96.  
  97.     public static void AsyncReleaseConfirmation() throws Exception {
  98.  
  99.         //获取信道
  100.         Channel channel = RabbitMQUtil.getChannel();
  101.         //开启发布确认
  102.         channel.confirmSelect();
  103.  
  104.         /**
  105.          * 线程安全有序的一个哈希表 设用于高并发的情况下
  106.          *1.能轻松的将序号与消息进行关联
  107.          * 2.可以轻松的批量删除内容，只要给到序号
  108.          * 3.支持高并发（多线程）
  109.          */
  110.         ConcurrentHashMap<Long, String> OutStandingConfirms = new ConcurrentHashMap<>();
  111.  
  112.         //随机生成队列名
  113.         String queueName = UUID.randomUUID().toString();
  114.         //队列构造 队列名/是否持久化/是否供多个消费者消费/是否自动删除/其他参数
  115.         channel.queueDeclare(queueName, true, false, false, null);
  116.  
  117.  
  118.  
  119.         //成功回调
  120.         /**
  121.          *
  122.          * deliveryTag 标记
  123.          * multiple 是否批量确认
  124.          */
  125.         ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
  126.             //2.判断时否是批量确认
  127.             if (multiple){
  128.                 //如果为批量确认，则直接清理
  129.                 OutStandingConfirms.clear();
  130.             }else {
  131.                 //若不为批量确认，则删除对应序号的消息
  132.                 OutStandingConfirms.remove(deliveryTag);
  133.             }
  134.  
  135.             System.out.println("确认成功的消息"+deliveryTag);
  136.         };
  137.         //失败回调
  138.         ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
  139.  
  140.             System.out.println("确认失败的消息"+deliveryTag);
  141.         };
  142.  
  143.         //开启监听
  144.         /**
  145.          *
  146.          * 1.监听成功的消息
  147.          * 2.监听失败的消息
  148.          */
  149.         channel.addConfirmListener(ackCallback, nackCallback);//异步通知
  150.  
  151.         //开始时间
  152.         long begin = System.currentTimeMillis();
  153.  
  154.         int counter = 0;
  155.         for (int i = 0; i < MAX; i++) {
  156.             String message = "消息" + i;
  157.  
  158.             //发送到哪个交换机/队列名/其他参数/发送消息的消息体
  159.             channel.basicPublish("", queueName, null, message.getBytes("UTF-8"));
  160.             //1:记录下所有要发送的消息     消息的总和
  161.             OutStandingConfirms.put(channel.getNextPublishSeqNo(),message);
  162.             counter++;
  163.         }
  164.  
  165.         if (counter == MAX) {
  166.             long end = System.currentTimeMillis();
  167.             long result = end - begin;
  168.             System.out.println("发送1000条消息使用了" + result + "ms");
  169.         }
  170.     }
  171. }

五、交换机

发布、订阅模式fanout        （广播：多个交换机都会收到）——效率最低

测试代码——生产者：

  1. public class EmitLog {
  2.  
  3.     public static final String EXCHANGE_NAME="logs" ;
  4.  
  5.     public static void main(String[] args) throws Exception {
  6.         //获取信道
  7.         Channel channel = RabbitMQUtil.getChannel();
  8.  
  9.         //构造交换机
  10.         channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
  11.  
  12.         //输入
  13.         Scanner scanner = new Scanner(System.in);
  14.  
  15.  
  16.         while (scanner.hasNext()){
  17.             String message = scanner.next();
  18.             channel.basicPublish(EXCHANGE_NAME,"123",null,message.getBytes("UTF-8"));
  19.  
  20.             System.out.print("发送消息"+message);
  21.         }
  22.     }
  23. }

测试代码——消费者1

  1. public class ReceiveLogs01 {
  2.  
  3.     //交换机名称
  4.     public static final String EXCHANGE_NAME="logs";
  5.  
  6.     public static void main(String[] args) throws Exception{
  7.  
  8.         Channel channel = RabbitMQUtil.getChannel();
  9.         //声明一个交换机
  10.         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
  11.  
  12.         //声明一个队列 临时队列
  13.         /**
  14.          * 队列名称是随机的
  15.          * 当消费者断开与队列的连接的时候 队列就自动删除
  16.          */
  17.         String queue = channel.queueDeclare().getQueue();
  18.  
  19.         /**
  20.          * 绑定交换机与队列
  21.          */
  22.         channel.queueBind(queue,EXCHANGE_NAME,"123");
  23.         System.out.println("ReceiveLogs01等待接收消息");
  24.  
  25.         //接收回调
  26.         DeliverCallback handle=(consumerTag, message)->{
  27.             System.out.println("ReceiveLogs01接收成功"+new String(message.getBody(),"UTF-8"));
  28.         };
  29.  
  30.         //失败回调
  31.         CancelCallback Cancel=(consumerTag)->{
  32.             System.out.println("ReceiveLogs01接收失败,序号"+consumerTag);
  33.         };
  34.  
  35.         channel.basicConsume(queue,true,handle,Cancel);
  36.  
  37.     }
  38. }

测试代码——消费者2 

  1. public class ReceiveLogs02 {
  2.  
  3.     //交换机名称
  4.     public static final String EXCHANGE_NAME = "logs";
  5.  
  6.     public static void main(String[] args) throws Exception {
  7.  
  8.         Channel channel = RabbitMQUtil.getChannel();
  9.         //声明一个交换机
  10.         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
  11.  
  12.         //声明一个队列 临时队列
  13.         /**
  14.          * 队列名称是随机的
  15.          * 当消费者断开与队列的连接的时候 队列就自动删除
  16.          */
  17.         String queue = channel.queueDeclare().getQueue();
  18.  
  19.         /**
  20.          * 绑定交换机与队列
  21.          */
  22.         channel.queueBind(queue, "logs", "123");
  23.         System.out.println("ReceiveLogs02等待接收消息");
  24.         
  25.         //接收回调
  26.         DeliverCallback handle = (consumerTag, message) -> {
  27.             System.out.println("ReceiveLogs02接收成功" + new String(message.getBody(), "UTF-8"));
  28.         };
  29.  
  30.         //失败回调
  31.         CancelCallback Cancel = (consumerTag) -> {
  32.             System.out.println("ReceiveLogs02接收失败,序号" + consumerTag);
  33.         };
  34.  
  35.         channel.basicConsume(queue, true, handle, Cancel);
  36.  
  37.     }
  38. }

直接交换机direct                 （竞争：如果有多个交换机，消息只会被发送给其中一个）——效率大大优于fanout，但是低于topics

代码与上面类似，仅需修改交换机的类型

主题交换机topics                 （指定：消息会发给对应routingKey的交换机，如果与多个交换机的routingKey符合，多个交换机都会收到消息）——效率最高

指定routingKey时：

*表示一个任意单词

#表示0-无数个单词

各个单词使用.分隔

测试代码同上，仅需修改routingKey与交换机类型

六、死信队列

死信产生的原因：

1.消息被拒绝

关闭自动应答

 channel.basicConsume(NORMAL_QUEUE,false,delivercallback,callback);

 修改成功回调

  1. //成功回调
  2.         DeliverCallback delivercallback=(consumerTag,message)->{
  3.             String msg = new String(message.getBody(), "UTF-8");
  4.             if (msg.equals("info5")){
  5.                 //拒绝应答，且不放回normal队列
  6.                 channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
  7.                 System.out.println("consumer01拒绝接收，消息为"+new String(message.getBody(),"UTF-8")+message.getEnvelope().getExchange());
  8.  
  9.             }else {
  10.                 //应答
  11.                 channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
  12.                 System.out.println("consumer01成功接收，消息为"+new String(message.getBody(),"UTF-8")+message.getEnvelope().getExchange());
  13.             }
  14.         };

2.ttl时间过期

ttl即 time to live 存活时间，可以通过消费者或者生产者设置，推荐使用生产者设置。

测试代码--生产者

  1. public class Producer {
  2.  
  3.     public static final String NORMAL_EXCHANGE = "normal_exchange";
  4.  
  5.     public static void main(String[] args) throws Exception {
  6.         //获取信道
  7.         Channel channel = RabbitMQUtil.getChannel();
  8.  
  9. //        声明存活时间(time to live)为10s
  10.         AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().expiration("10000").build();
  11.  
  12.         for (int i = 0; i < 11; i++) {
  13.             //发送的消息
  14.             String message = "info" + i;
  15.             //推送消息
  16.             channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", props, message.getBytes());
  17.         }
  18.  
  19.     }
  20. }

 测试代码--NORMAL消费者

  1. public class Consumer01 {
  2.     //声明交换机名
  3.     public static final String NORMAL_EXCHANGE="normal_exchange";
  4.     public static final String DEAD_EXCHANGE="dead_exchange";
  5.  
  6.     //声明队列名
  7.     public static final String NORMAL_QUEUE="normal_queue";
  8.     public static final String DEAD_QUEUE="dead_queue";
  9.  
  10.     public static void main(String[] args) throws Exception{
  11.  
  12.         Channel channel = RabbitMQUtil.getChannel();
  13.         //声明死信与普通交换机
  14.         channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
  15.         channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
  16.         //声明普通队列
  17.         HashMap<String, Object> arguments = new HashMap<>();
  18.  
  19.         //设置ttl时间为10s
  20.         arguments.put("x-message-ttl",10000);
  21.         //设置死信交换机
  22.         arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
  23.         //设置死信交换机的routingKey
  24.         arguments.put("x-dead-letter-routing-key","lisi");
  25.  
  26.         channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
  27.  
  28.         //声明死信队列
  29.         channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
  30.  
  31.         //绑定交换机
  32.         channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
  33.         channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
  34.  
  35.         //成功回调
  36.         DeliverCallback delivercallback=(consumerTag,message)->{
  37.             System.out.println("consumer01成功接收，消息为"+new String(message.getBody(),"UTF-8")+message.getEnvelope().getExchange());
  38.         };
  39.         //失败回调
  40.         CancelCallback callback=(consumerTag)->{
  41.             System.out.println("接收失败,标记为"+consumerTag);
  42.         };
  43.         channel.basicConsume(NORMAL_QUEUE,delivercallback,callback);
  44.     }
  45. }

测试代码--DEAD消费者

  1. public class DeadConsumer {
  2.     
  3.     public static final String DEAD_QUEUE="dead_queue";
  4.  
  5.     public static void main(String[] args) throws Exception {
  6.         Channel channel = RabbitMQUtil.getChannel();
  7.  
  8.  
  9.         //成功回调
  10.         DeliverCallback delivercallback=(consumerTag, message)->{
  11.             System.out.println("deadConsumer成功接收，消息为"+new String(message.getBody(),"UTF-8")+message.getEnvelope().getExchange());
  12.         };
  13.         //失败回调
  14.         CancelCallback callback=(consumerTag)->{
  15.             System.out.println("接收失败,标记为"+consumerTag);
  16.         };
  17.         channel.basicConsume(DEAD_QUEUE,delivercallback,callback);
  18.     }
  19. }

3.超过队列最大长度

测试代码如上，仅需在设置中增加一个队列长度

  1.         //设置队列NORMAL_QUEUE的最大长度为6 注意该处第二个参数不是string类型
  2.         arguments.put("x-max-length",6);

注意，在完成上面的测试后，如想修改配置并测试本项，需将之前生成的队列删除，以避免报错

七、延迟队列

1、利用死信队列来实现延迟

新建一个springboot项目，引入如下依赖

  1.  <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
  2.         <dependency>
  3.             <groupId>org.springframework.boot</groupId>
  4.             <artifactId>spring-boot-starter-amqp</artifactId>
  5.         </dependency>
  6.  
  7.         <!--swagger-->
  8.         <!-- https://mvnrepository.com/artifact/com.github.xiaoymin/knife4j-spring-boot-starter -->
  9.         <dependency>
  10.             <groupId>com.github.xiaoymin</groupId>
  11.             <artifactId>knife4j-spring-boot-starter</artifactId>
  12.             <version>3.0.3</version>
  13.         </dependency>
  14.         
  15.         <!--swagger-ui-->
  16.         <dependency>
  17.             <groupId>io.springfox</groupId>
  18.             <artifactId>springfox-swagger-ui</artifactId>
  19.             <version>3.0.0</version>
  20.             <scope>compile</scope>
  21.         </dependency>
  22.  
  23.         <!--json-->
  24.         <!-- https://mvnrepository.com/artifact/com.alibaba.fastjson2/fastjson2 -->
  25.         <dependency>
  26.             <groupId>com.alibaba.fastjson2</groupId>
  27.             <artifactId>fastjson2</artifactId>
  28.             <version>2.0.23</version>
  29.         </dependency>
  30.  
  31.         <!-- https://mvnrepository.com/artifact/org.springframework.amqp/spring-rabbit-test -->
  32.         <dependency>
  33.             <groupId>org.springframework.amqp</groupId>
  34.             <artifactId>spring-rabbit-test</artifactId>
  35.             <version>3.0.8</version>
  36.             <scope>test</scope>
  37.         </dependency>
  38.         
  39.         <dependency>
  40.             <groupId>org.springframework.boot</groupId>
  41.             <artifactId>spring-boot-starter-thymeleaf</artifactId>
  42.         </dependency>
  43.         <dependency>
  44.             <groupId>org.springframework.boot</groupId>
  45.             <artifactId>spring-boot-starter-web</artifactId>
  46.         </dependency>
  47.  
  48.         <dependency>
  49.             <groupId>org.springframework.boot</groupId>
  50.             <artifactId>spring-boot-devtools</artifactId>
  51.             <scope>runtime</scope>
  52.             <optional>true</optional>
  53.         </dependency>
  54.         <dependency>
  55.             <groupId>org.projectlombok</groupId>
  56.             <artifactId>lombok</artifactId>
  57.             <optional>true</optional>
  58.         </dependency>
  59.         <dependency>
  60.             <groupId>org.springframework.boot</groupId>
  61.             <artifactId>spring-boot-starter-test</artifactId>
  62.             <scope>test</scope>
  63.         </dependency>

 编写swagger config

  1. @Configuration
  2. @EnableSwagger2
  3. public class SwaggerConfig {
  4.  
  5.     public Docket webApiConfig(){
  6.  
  7.         return new Docket(DocumentationType.SWAGGER_2)
  8.                 .groupName("webApi")
  9.                 .apiInfo(webApiInfo())
  10.                 .select()
  11.                 .build();
  12.     }
  13.  
  14.     private ApiInfo webApiInfo(){
  15.  
  16.         return new ApiInfoBuilder()
  17.                 .title("rabbitmq接口文档")
  18.                 .description("本文档描述了rabbitmq微服务接口定义")
  19.                 .version("1.0")
  20.                 .contact(new Contact("WangXu","http://swagger.com","xu.wangcn@foxmail.com"))
  21.                 .build();
  22.     }
  23. }

 声明队列、交换机并绑定

  1. @Component
  2. public class TtlQueueConfig {
  3.  
  4.     //普通交换机名
  5.     public static final String X_EXCHANGE="X";
  6.     //死信交换机名
  7.     public static final String Y_DEAD_LETTER_EXCHANGE="Y";
  8.     //普通队列名
  9.     public static final String QUEUE_A="QA";
  10.     public static final String QUEUE_B="QB";
  11.     //死信队列名
  12.     public static final String DEAD_LETTER_QUEUE="QD";
  13.  
  14.  
  15.     //声明xExchange
  16.     @Bean("xExchange")
  17.     public DirectExchange xExchange(){
  18.         return new DirectExchange(X_EXCHANGE);
  19.     }
  20.     //声明xExchange
  21.     @Bean("yExchange")
  22.     public DirectExchange yExchange(){
  23.         return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
  24.     }
  25.  
  26.     //声明队列A
  27.     @Bean("queueA")
  28.     public Queue aQueue(){
  29.         Map<String, Object> arguments=new HashMap<>();
  30.         //设置死信交换机
  31.         arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
  32.         //设置死信routing key
  33.         arguments.put("x-dead-letter-routing-key","YD");
  34.         //设置消息过期时间
  35.         arguments.put("x-message-ttl",10000);
  36.         return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
  37.     }
  38.  
  39.     //声明队列B
  40.     @Bean("queueB")
  41.     public Queue bQueue(){
  42.         Map<String, Object> arguments=new HashMap<>();
  43.         //设置死信交换机
  44.         arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
  45.         //设置死信routing key
  46.         arguments.put("x-dead-letter-routing-key","YD");
  47.         //设置消息过期时间
  48.         arguments.put("x-message-ttl",40000);
  49.         return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
  50.     }
  51.  
  52.     //声明队列D
  53.     @Bean("queueD")
  54.     public Queue dQueue(){
  55.         return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
  56.     }
  57.  
  58.     //绑定交换机与队列
  59.     @Bean
  60.     public Binding xExchangeBindingQaQueue(@Qualifier("queueA") Queue queueA,@Qualifier("xExchange") DirectExchange xExchange){
  61.         return BindingBuilder.bind(queueA).to(xExchange).with("XA");
  62.     }
  63.  
  64.     //绑定交换机与队列
  65.     @Bean
  66.     public Binding xExchangeBindingQbQueue(@Qualifier("queueB") Queue queueB,@Qualifier("xExchange") DirectExchange xExchange){
  67.         return BindingBuilder.bind(queueB).to(xExchange).with("XB");
  68.     }
  69.  
  70.     //绑定死信交换机与死信队列
  71.     @Bean
  72.     public Binding yExchangeBindingQdQueue(@Qualifier("queueD") Queue queueD,@Qualifier("yExchange") DirectExchange yExchange){
  73.         return BindingBuilder.bind(queueD).to(yExchange).with("YD");
  74.     }
  75. }

 编写死信消费者

  1. @Slf4j
  2. @Component
  3. public class DeadLetterQueueConsumer {
  4.  
  5.     //接收消息
  6.     @RabbitListener(queues="QD")
  7.     public void receiveD(Message message, Channel channel) throws Exception{
  8.         String msg=new String(message.getBody(),"UTF-8");
  9.         log.info("当前时间为：{}，接收到的消息为{}",new Date().toString(),msg);
  10.     }
  11. }

 配置rabbitmq参数

  1. spring.rabbitmq.host=172.20.10.2
  2. spring.rabbitmq.username=admin
  3. spring.rabbitmq.password=root
  4. spring.mvc.pathmatch.matching-strategy=ant_path_matcher

消费者 & controller

  1. @Slf4j
  2. @RestController()
  3. @RequestMapping("/ttl")
  4. public class SendMsgController {
  5.  
  6.     @Autowired
  7.     private RabbitTemplate rt;
  8.  
  9.     @GetMapping("/sendMsg/{message}")
  10.     public void postMsg(@PathVariable("message") String message) {
  11.         log.info("当前时间:{},发送一条消息给两个TTL队列：{}", new Date().toString(), message);
  12.         rt.convertAndSend("X", "XA", "消息来自ttl为10s的交换机" + message);
  13.         rt.convertAndSend("X", "XB", "消息来自ttl为40s的交换机" + message);
  14.  
  15.     }
  16. }

swagger后台访问地址

Swagger各版本访问地址:

2.9.x 访问地址:
http://ip:port/{context-path}/swagger-ui.html

3.0.x 访问地址:
http://ip:port/{context-path}/swagger-ui/index.html

3.0集成knife4j 访问地址:
http://ip:port/{context-path}/doc.html

注意点：

Channel需要引用import com.rabbitmq.client.Channel;包

测试结果 

使用死信队列的方式实现延迟虽然实现了，但是它有一个问题，该方式只能顺序消费消息，当第一个消息的ttl时间大于后面消息的ttl时间时，延迟将为第一个消息的ttl

 2、利用插件实现延迟

 于是，为了解决这个问题，我们就需要用到rabbitmq的延迟队列插件了

 首先我们先前往github上下载延迟队列的插件

Releases · rabbitmq/rabbitmq-delayed-message-exchange · GitHub

随后将下载的插件放到linux服务器中

scp "C:\Users\q\Downloads\rabbitmq_delayed_message_exchange-3.8.0.ez" root@172.20.10.2:/root/rabbit

 复制插件到rabbitmq插件目录下

cp rabbitmq_delayed_message_exchange-3.8.0.ez /usr/lib/rabbitmq/lib/rabbimtq_server-3.8.6/plugins

 安装插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

 重启rabbitmq服务

systemctl restart rabbitmq-server

 打开后台查看交换机类型是否多出一种x_delayed-message,如果有那就是成功了

测试通过插件来实现延迟队列

编写配置类

  1. @Configuration
  2. public class DelayedQueueConfig {
  3.  
  4.     //交换机名称
  5.     public static final String DELAYED_EXCHANGE_NAME="delayed.exchange";
  6.     //队列名称
  7.     public static final String DELAYED_QUEUE_NAME="delayed.queue";
  8.     //routing key
  9.     public static final String DELAYED_ROUTING_KEY_NAME="delayed.routing.key";
  10.  
  11.     //声明队列
  12.     @Bean("dQueue")
  13.     public Queue getQueue(){
  14.         return QueueBuilder.durable(DELAYED_QUEUE_NAME).build();
  15.     }
  16.  
  17.     //声明交换机
  18.     @Bean("dExchange")
  19.     public CustomExchange getExchange(){
  20.         HashMap<String, Object> arguments = new HashMap<>();
  21.         arguments.put("x-delayed-type","direct");
  22.         /**
  23.          * 1.交换机名称
  24.          * 2.交换机类型
  25.          * 3.是否需要持久化
  26.          * 4.是否自动删除
  27.          * 5.其他参数
  28.          */
  29.         return new CustomExchange(DELAYED_EXCHANGE_NAME,"x-delayed-message",false,false,arguments);
  30.     }
  31.  
  32.     //声明绑定关系
  33.     @Bean
  34.     public Binding dQueueBindingD(@Qualifier("dQueue")Queue dQueue, @Qualifier("dExchange")CustomExchange dExchange){
  35.         return BindingBuilder.bind(dQueue).to(dExchange).with(DELAYED_ROUTING_KEY_NAME).noargs();
  36.     }
  37. }

在controller类中编写请求

  1.     //编写get请求
  2.     @GetMapping("/delayed/{message}/{delayedTime}")
  3.     public void testTwo(@PathVariable("message") String message, @PathVariable("delayedTime") Integer delayedTime) {
  4.         //控制台打印消息
  5.         log.info("{}发送了一个消息：{},delayedTime:{}", new Date(), message,delayedTime);
  6.         //指明消息要发送的交换机及routing key，并设置延迟时间
  7.         rt.convertAndSend("delayed.exchange", "delayed.routing.key", message,msg->{
  8.             msg.getMessageProperties().setDelay(delayedTime);
  9.             return msg;
  10.         });
  11.     }

编写消费者

  1. @Slf4j
  2. @Component
  3. public class DelayedQueueConsumer {
  4.  
  5.     //接收消息
  6.     @RabbitListener(queues="delayed.queue")
  7.     public void receiveD(Message message, Channel channel) throws Exception{
  8.         String msg=new String(message.getBody(),"UTF-8");
  9.         log.info("延迟队列当前时间为：{}，接收到的消息为{}",new Date().toString(),msg);
  10.     }
  11. }

 测试结果

 总结

延时队列在需要延时处理的场景下非常有用，延迟队列可以有效地利用Rabbitmq的特性，如消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次，以及未确认处理的消息不会被丢弃。另外，通过RabbitMQ集群的特性，可以很好的解决单点故障问题，不会因为单个接待你挂掉导致延时队列不可用或者消息丢失。

当然，延时队列还有其它选择，比如java的DelayQueue，利用Redis的zset，利用quartz或者利用kafka的时间轮，这些方式各有特点，看需要适用的场景。

八、发布确认高级内容

在生产环境中由于一些不明原因导致rabbitmq重启，在rabbitmq重启期间生产者消息投递失败导致消息丢失，需要手动处理和恢复。于是，我们开始思考，如何才能进行rabbitmq的消息可靠投递呢？特别是在这样比较极端的情况，rabbitmq集群不可用的时候，无法投递的消息该如何处理呢？

 1、确认回调接口

编写测试用例：当交换机没有接收到消息时调用confirmCallBack

配置类

  1. //开启日志支持
  2. @Slf4j
  3. //将该类注入容器
  4. @Component
  5. public class ConfirmConfig {
  6.  
  7.     //声明交换机名
  8.     public static final String EXCHANGE_NAME="confirm.exchange";
  9.     //声明队列名
  10.     public static final String QUEUE_NAME="confirm.queue";
  11.     //声明routing key
  12.     public static final String ROUTING_KEY="key1";
  13.  
  14.     @Bean("confirmExchange")
  15.     public DirectExchange confirmExchange(){
  16.         //构建交换机
  17.         return new DirectExchange(EXCHANGE_NAME,true,false);
  18.     }
  19.  
  20.     @Bean("confirmQueue")
  21.     public Queue confirmQueue(){
  22.         //构建队列
  23.         return QueueBuilder.durable(QUEUE_NAME).build();
  24.     }
  25.  
  26.     @Bean
  27.     public Binding qBindingExchange(@Qualifier("confirmExchange")DirectExchange confirmExchange,
  28.                                     @Qualifier("confirmQueue")Queue confirmQueue){
  29.         //绑定队列、交换机、routing key
  30.         return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(ROUTING_KEY);
  31.     }
  32. }

 生产者

  1.     @GetMapping("/confirm/{message}")
  2.     public void confirmTest(@PathVariable("message") String message) {
  3.         CorrelationData correlationData = new CorrelationData("1");
  4.         log.info("当前时间:{},发送一条消息：{}", new Date().toString(), message);
  5.         rt.convertAndSend(ConfirmConfig.EXCHANGE_NAME, ConfirmConfig.ROUTING_KEY, "消息:" +
  6.                 message,correlationData);
  7.  
  8.     }

消费者

  1. @Slf4j
  2. @Component
  3. public class ConfirmConsumer {
  4.  
  5.     @RabbitListener(queues = "confirm.queue")
  6.     public void receiveQ(Message message, Channel channel) throws UnsupportedEncodingException {
  7.         String msg=new String(message.getBody(),"UTF-8");
  8.         log.info("已经收到消息：{},当前时间为：{}",msg,new Date());
  9.     }
  10. }

编写confirmCallBack实现类，实现消息不管是否到达交换机都会有回调

  1. @Slf4j
  2. @Component
  3. public class MyCallBack implements RabbitTemplate.ConfirmCallback {
  4.  
  5.     //如果我们实现的是一个类的内部接口，而当我们通过该类调用该接口时无法调到我们写的实现类，
  6.     // 所以要先注入该类
  7.     @Autowired
  8.     private RabbitTemplate rabbitTemplate;
  9.  
  10.     @PostConstruct
  11.     public void setConfirmCallback(){
  12.         //将该实现类设置为rabbitTemplate的确认回调
  13.         rabbitTemplate.setConfirmCallback(this);
  14.     }
  15.  
  16.  
  17.     /**
  18.      * @param correlationData correlation data for the callback.
  19.      * @param ack true for ack, false for nack
  20.      * @param cause An optional cause, for nack, when available, otherwise null.
  21.      * 交换机确认回调方法
  22.      * 1.发消息 交换机收到了 回调
  23.      *  1.1 correlationData 保存回调消息的id及相关信息
  24.      *  1.2 交换机收到了消息 true
  25.      *  1.3 cause null
  26.      *
  27.      * 2.发消息 交换机没收到 回调
  28.      *  2.1 correlationData 保存回调消息的id及相关信息
  29.      *  2.2 交换机收到了消息 false
  30.      *  2.3 cause 失败的原因
  31.      */
  32.     @Override
  33.     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
  34.  
  35.         if(ack){
  36.             log.info("成功应答，消息id为：{}",correlationData.getId());
  37.         }else {
  38.             log.info("交换机没有收到消息，消息的id为{},失败的原因为：{}",correlationData.getId(),cause);
  39.         }
  40.     }
  41. }

配置确认类型

  1. #默认是none 禁止发布确认模式 correlated 发布消息成功到交换机后会触发回调方法
  2. #simple 有两种效果，其一与correlated一样会触发回调方法，其二根据返回结果来判定下一步的逻辑
  3. spring.rabbitmq.publisher-confirm-type=correlated

 在生产者处将交换机名写错，测试结果如下：

 2、回退消息

当交换机可以正常接收，而队列无法接收时我们需要让消息回退

首先我们需要在MyCallBack中实现回退接口

  1. @Slf4j
  2. @Component
  3. public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
  4.  
  5.     //如果我们实现的是一个类的内部接口，而当我们通过该类调用该接口时无法调到我们写的实现类，
  6.     // 所以要先注入该类
  7.     @Autowired
  8.     private RabbitTemplate rabbitTemplate;
  9.  
  10.     @PostConstruct
  11.     public void setConfirmCallback(){
  12.         //将该实现类设置为rabbitTemplate的确认回调
  13.         rabbitTemplate.setConfirmCallback(this);
  14.     }
  15.  
  16.     @PostConstruct
  17.     public void setReturnsCallBack(){
  18.         //将该实现类设置为rabbitTemplate的回退方法
  19.         rabbitTemplate.setReturnsCallback(this);
  20.     }
  21.  
  22.  
  23.     /**
  24.      * @param correlationData correlation data for the callback.
  25.      * @param ack true for ack, false for nack
  26.      * @param cause An optional cause, for nack, when available, otherwise null.
  27.      * 交换机确认回调方法
  28.      * 1.发消息 交换机收到了 回调
  29.      *  1.1 correlationData 保存回调消息的id及相关信息
  30.      *  1.2 交换机收到了消息 true
  31.      *  1.3 cause null
  32.      *
  33.      * 2.发消息 交换机没收到 回调
  34.      *  2.1 correlationData 保存回调消息的id及相关信息
  35.      *  2.2 交换机收到了消息 false
  36.      *  2.3 cause 失败的原因
  38.      */
  39.     @Override
  40.     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
  41.  
  42.         if(ack){
  43.             log.info("交换机成功应答，消息id为：{}",correlationData.getId());
  44.         }else {
  45.             log.info("交换机没有收到消息，消息的id为{},失败的原因为：{}",correlationData.getId(),cause);
  46.         }
  47.     }
  48.  
  49.     //回退方法，可以在消息传递过程中不可达目的地时将消息返回给生产者
  50.     //只有消息不可达时，才会回退
  51.     @Override
  52.     public void returnedMessage(ReturnedMessage returned) {
  53.         try {
  54.             log.error("消息{}，被交换机{}退回，退回原因：{}，路由Key：{}",new String(returned.getMessage().getBody(),"UTF-8"),
  55.                     returned.getExchange(), returned.getReplyText(),returned.getRoutingKey());
  56.         } catch (UnsupportedEncodingException e) {
  57.             throw new RuntimeException(e);
  58.         }
  59.     }

 修改配置类，新增

spring.rabbitmq.publisher-returns=true

在生产者中将routing key修改为错误的，并进行测试

3、备份交换机

声明交换机和队列并绑定

  1.  //备份交换机
  2.     @Bean("backupExchange")
  3.     public FanoutExchange backupExchange(){
  4.         return new FanoutExchange(BACKUP_EXCHANGE_NAME,true,false);
  5.     }
  6.  
  7.     //备份队列
  8.     @Bean("backupQueue")
  9.     public Queue backupQueue(){
  10.         return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
  11.     }
  12.  
  13.     //警告队列
  14.     @Bean("warningQueue")
  15.     public Queue waringQueue(){
  16.         return QueueBuilder.durable(WARING_QUEUE_NAME).build();
  17.     }
  18.  
  19.     //绑定交换机与队列
  20.     @Bean
  21.     public Binding backupBindingBackupExchange(@Qualifier("backupQueue")Queue backupQueue,
  22.                                                @Qualifier("backupExchange")FanoutExchange backupExchange ){
  23.         return BindingBuilder.bind(backupQueue).to(backupExchange);
  24.     }
  25.  
  26.     //绑定交换机与队列
  27.     @Bean
  28.     public Binding waringBindingBackupExchange(@Qualifier("warningQueue")Queue warningQueue,
  29.                                                @Qualifier("backupExchange")FanoutExchange backupExchange ){
  30.         return BindingBuilder.bind(warningQueue).to(backupExchange);
  31.     }

 在确认交换机配置备份交换机

  1.     @Bean("confirmExchange")
  2.     public DirectExchange confirmExchange(){
  3.         //构建交换机
  4.         return ExchangeBuilder.directExchange(EXCHANGE_NAME).durable(true)
  5.         //配置备份交换机
  6.                 .withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
  7.     }

消费者

  1. @Slf4j
  2. @Component
  3. public class warningConsumer {
  4.  
  5.     @RabbitListener(queues= ConfirmConfig.WARING_QUEUE_NAME)
  6.     public void receiveD(Message message, Channel channel) throws UnsupportedEncodingException {
  7.         String msg = new String(message.getBody(), "UTF-8");
  8.         log.info("warningQueue已收到消息：{}，时间为：{}",msg,new Date());
  9.  
  10.     }
  11.  
  12. }

 生产者

  1.     @GetMapping("/confirm/{message}")
  2.     public void confirmTest(@PathVariable("message") String message) {
  3.         CorrelationData correlationData = new CorrelationData("1");
  4.         log.info("当前时间:{},发送一条消息：{}", new Date().toString(), message);
  5.         rt.convertAndSend(ConfirmConfig.EXCHANGE_NAME, ConfirmConfig.ROUTING_KEY+"1", "消息:" +
  6.                 message,correlationData);
  7.  
  8.     }

回退消息与备份交换机同时存在，备份交换机优先级更高

  测试结果：

 九、幂等性

用户对于同一操作发起的一次请求或多次请求的结果时一致的，不会因为多次点击而产生副作用。举个例子，支付，有时候因为网络异常，已经扣了钱，但当用户再次点击按钮，此时会进行第二次扣款，最终用户查询余额变少了，流水记录也变成了两条。在以前的但应用系统中，我们只需要把数据放入事务中即可，发生错误立即回滚，但是再响应客户端的时候也有可能出现网络中断或者异常等等。

消息被重复消费

解决思路

使用全局ID或者写个唯一标识比如时间戳或者UUID或者订单消费者消费MQ中的消息也可利用MQ的该id来判断，或者可按自己的规则生成一个全局唯一的id，每次消费前用该id先判断消息是否已消费过。

消息永远不会被消费多次，业界主流的幂等性有两种操作，唯一id+指纹码机制，利用数据库主键去重，b利用redis的原子性去实现

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

十、优先级队列

十一、惰性队列

消息保存在内存还是磁盘中，正常情况下消息是保存在内存中，而惰性队列消息时保存在磁盘中。

当消费者由于各种各样的原因（比如消费者下线、宕机亦或者由于维护而关闭等）而使长时间内不能消费消息造成堆积时，惰性队列就很有必要了。

在队列声明时可以通过“x-queue-mode”参数来设置队列的模式，取值为“default”和“lazy”。

 十二、rabbitmq集群

1.克隆两台虚拟机并启动
2.修改机器的主机名称

vim /etc/hostname

三台机器分别命名为node1、node2、node3

3.配置各个节点的hosts文件，让各个节点都能互相识别对方

vim /etc/hosts

ip1 node1

ip2 node2

ip3 node3

4.确保各个节点使用的cookie文件使用的是同一个值

在node1上执行远程操作命令

scp /var/lib/rabbitmq/.erlang.cookie root@node2:/var/lib/rabbitmq/.erlang.cookie

scp /var/lib/rabbitmq/.erlang.cookie root@node3:/var/lib/rabbitmq/.erlang.cookie

5.启动RabbitMQ服务，顺带启动Erlang虚拟机和RabbitMQ应用服务（在三台节点上分别执行）

rabbitmq-server -detached

6.在克隆的两台虚拟机中执行如下命令：

rabbitmqctl stop_app

(rabbitmqctl stop会将Erlang虚拟机关闭,rabbitmqctl stop_app只关闭RabbitMQ服务)

rabbitmqctl reset

rabbitmqctl join_cluster rabbit@node1/第三台机器也可以执行：rabbitmqctl join_cluster rabbit@node2

rabbitmqctl start_app(只启动应用服务)

7.查看集群状态

rabbitmqctl cluster_status

8.创建集群用户账号

创建账号：rabbitmqctl add_user admin 123

设置角色：rabbitmqctl set_user_tags admin administrator

设置权限：rabbitmqctl set_permissions -p "/" admin "." ".*" ".*"

9.解除集群节点

rabbitmqctl stop_app

rabbitmqctl reset

rabbitmqctl start_app

rabbitmqctl cluster_status

rabbitmqctl forget_cluster_node rabbit@node2(node1机器上执行)

十三、镜像队列

如果RabbitMQ集群中只有一个Broker节点，那么该节点的失效将导致整体服务的临时性不可用，并且也可能会导致消息的丢失。可以将所有消息都设置持久化，并且对应队列的durable属性也设置为true，单消息发送之后和被写入磁盘并执行刷盘动作之间存在一个短暂却会产生问题的时间窗。通过publisherconfirm机制能够确保客户端知道哪些消息已经存入磁盘，尽管如此，一般不希望遇到因单点故障导致的服务不可用。

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

打开rabbitmq后台选择Admin-->policies

Haproxy高可用负载均衡

当我们有海量的请求时，如果我们仍然将请求发送到同一个服务器上时，服务器可能会崩溃，而有些服务器可能会因为没有收到相应而空闲，这时我们就需要用到Haproxy来帮助我们均衡的分配请求，从而使用户获得良好的体验，同时也避免服务器因大量请求而崩溃。

Federation Exchange联邦交换机

当我们的用户来自有不同的地区时，遥远的用户可能会因网络延时而体验变差，为了用户能够在不同地区体验到大致相同的体验，我们可以在不同的地区同时部署服务器。

而当我们在多个不同的地点部署服务器后，我们需要解决各个服务器的数据一致问题，这时我们就需要用到联邦交换机。

搭建步骤：

1.需要保证每台节点单独运行

2.哎每台机器上开启federation相关插件

rabbitmq-plugins enable rabbitmq_federtion

rabbitmq-plugins enable rabbitmq_federation_management

在从节点downstream（node2）上创建一个名为“fed_exchange”的交换机并声明队列

在downstream（node2）配置upstream（node1）

admin-->Federation Upstreams

Federation Queue

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

搭建步骤

Shovel

首先在服务器中开启shovel支持

rabbitmq plugins enable rabbitmq_shovel

rabbitmq plugins enable rabbitmq_shovel_management

然后打开rabbitmq管理后台admin-->shovel management

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

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

  

闽ICP备14008679号