定制软件开发【学相伴】狂神说 RabbitMQ笔记（简单使用RabbitMQ）

目录

什么是

 RabbitMQ是一个由erlang开发的AMQP（Advanced Message Queue 定制软件开发高级消息队列协议 ）定制软件开发的开源实现，定制软件开发能够实现定制软件开发异步消息处理
 RabbitMQ定制软件开发是一个消息代理：定制软件开发它接受和转发消息。
定制软件开发你可以把它想象成一个邮局：定制软件开发当你把你想要发布的邮定制软件开发件放在邮箱中时，定制软件开发你可以确定邮差先生最定制软件开发终将邮件发送给你的收件人。定制软件开发在这个比喻中，RabbitMQ定制软件开发是邮政信箱，定制软件开发邮局和邮递员。 
RabbitMQ定制软件开发和邮局的主要区别在于定制软件开发它不处理纸张，而是接受，定制软件开发存储和转发二进制数据块
        
优点：异步消息处理
           业务解耦（定制软件开发下订单操作：扣减库存、生成订单、发红包、发短信），定制软件开发将下单操作主流程：扣减库存、定制软件开发生成订单然后通过MQ定制软件开发消息队列完成通知，发红包、发短信
            错峰流控 （通知量 消息量 定制软件开发订单量大的情况实现MQ定制软件开发消息队列机制，定制软件开发淡季情况下访问量会少）
        
              定制软件开发灵活的路由（Flexible Routing） 
             定制软件开发在消息进入队列之前，通过 Exchange 定制软件开发来路由消息的。定制软件开发对于典型的路由功能，RabbitMQ 定制软件开发已经提供了一些内置的 Exchange 来实现。定制软件开发针对更复杂的路由功能，定制软件开发可以将多个 Exchange 定制软件开发绑定在一起，定制软件开发也通过插件机制实现自己的 Exchange 。
        RabbitMQ定制软件开发网站端口号：15672
        定制软件开发程序里面实现的端口为：5672

       

 

使用docker安装RabbitMQ,定制软件开发如果没有使用过的可以定制软件开发看这篇文章

1.拉取RabbitMQ镜像

docker pull rabbitmq:management

2.运行RabbitMQ镜像

docker run -itd --name rabbit01 --hostname myrabbit -e RABBITMQ_DEFAULT_USER=chenjinxian -e RABBITMQ_DEFAULT_PASS=chenjinxian -p 15672:15672 -p  5672:5672 -p 25672:25672 rabbitmq:management

注意：RABBITMQ_DEFAULT_USER=chenjinxian -e RABBITMQ_DEFAULT_PASS=chenjinxian 

   定制软件开发这里设置的是（RABBITMQ_DEFAULT_USER）定制软件开发登录的账号和（ RABBITMQ_DEFAULT_PASS）密码，定制软件开发根据自身来修改

 定制软件开发这里看到容器已经开启成功了，定制软件开发然后就可以使用了

 

3.定制软件开发通过浏览器打开

定制软件开发如果你使用的是本地虚拟机，定制软件开发那么你直接使用虚拟机显示的ipv4定制软件开发地址加端口号就可以访问了；

定制软件开发如果你使用的是云服务器，定制软件开发那么你需要在对应服务器（阿里云，腾讯云等）定制软件开发的安全组中开放15672端口，定制软件开发并且在防火墙中也开放15672端口

 

 定制软件开发显示如上图那么就可以定制软件开发开始使用了

定制软件开发然后通过命令进入rabbitmq容器

docker exec -it rabbit01 /bin/bash

定制软件开发授权账号和密码

rabbitmqctl add_user admin admin

定制软件开发设置用户分配操作权限

rabbitmqctl set_user_tags admin administrator

用户级别：

  1. administrator：定制软件开发可以登录控制台、查看所有信息、可以对 rabbitmq进行管理

  2. monitoring：监控者 登录控制台，查看所有信息

  3. policymaker：策略制定者 登录控制台，指定策略

  4. managment 普通管理员 登录控制台

为用户添加资源权限 

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

也可以在界面操作进行添加用户

 

RabbitMQ支持的消息模型

1.简单模式 Simple  

 

     2.工作模式 Work

       

 

     3.发布订阅模式

 4.路由模式

5.主题 Topic模式

 

      6.参数模式

7.出版商确认模式

 

1.入门案例

1. RabbitMQ入门案例 - Simple 简单模式

  1. jdk1.8

  2. 构建一个 maven工程

  3. 定义生产者

  4. 定义消费者

  5. 观察消息的在 rabbitmq-server服务中的进程

 

01 构建一个maven工程

02 导入依赖

  1.  <dependencies>
  2.         <!--导入rabbitmq的依赖-->
  3.         <dependency>
  4.             <groupId>com.rabbitmq</groupId>
  5.             <artifactId>amqp-client</artifactId>
  6.             <version>5.13.0</version>
  7.         </dependency>
  8.  
  9.     </dependencies>

3.代码编写 

 在上图的模型中，有以下概念：

  1. 生产者，也就是要发送消息的程序

  2. 消费者：消息的接受者，会一直等待消息到来。

  3. 消息队列：图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

生产者 

  1. package com.chen.rabbitmq.simple;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6. import com.rabbitmq.client.MessageProperties;
  7.  
  8. /**
  9.  * @description: 简单模式Simple
  10.  */
  11. public class Producer {
  12.  
  13.  
  14.     public static void main(String[] args) {
  15.  
  16.         // 所有的中间件技术都是基于tcp/ip协议基础之上构建新型的协议规范，只不过rabbitmq遵循的是amqp
  17.         // ip port
  18.  
  19.         // 1: 创建连接工程
  20.         ConnectionFactory connectionFactory = new ConnectionFactory();
  21.         connectionFactory.setHost("128.197.157.151");
  22.         connectionFactory.setPort(5672);
  23.         connectionFactory.setUsername("chenjinxian");//rabbitmq登录的账号
  24.         connectionFactory.setPassword("chenjinxian");//rabbitmq登录的密码
  25.         connectionFactory.setVirtualHost("/");
  26.  
  27.         //springboot ---rabbitmq
  28.  
  29.         Connection connection = null;
  30.         Channel channel = null;
  31.         try {
  32.             // 2: 创建连接Connection Rabbitmq为什么是基于channel去处理而不是链接? 长连接----信道channel
  33.             connection = connectionFactory.newConnection("生成者");
  34.             // 3: 通过连接获取通道Channel
  35.             channel = connection.createChannel();
  36.             // 4: 通过通创建交换机，声明队列，绑定关系，路由key,发送消息，和接收消息
  37.             String queueName = "queue1";
  38.  
  39.             /*
  40.              * @params1 队列的名称
  41.              * @params2 是否要持久化durable=false 所谓持久化消息是否存盘，如果false 非持久化 true是持久化? 非持久化会存盘吗？ 会存盘，但是会随从重启服务会丢失。
  42.              * @params3 排他性，是否是独占独立
  43.              * @params4 是否自动删除，随着最后一个消费者消息完毕消息以后是否把队列自动删除
  44.              * @params5 携带附属参数
  45.              */
  46.             channel.queueDeclare(queueName, true, false, false, null);
  47.             // 5: 准备消息内容
  48.             String message = "Hello chenjinxian!!!";
  49.             // 6: 发送消息给队列queue
  50.             // @params1: 交换机  @params2 队列、路由key @params 消息的状态控制  @params4 消息主题
  51.             // 面试题：可以存在没有交换机的队列吗？不可能，虽然没有指定交换机但是一定会存在一个默认的交换机。
  52.             channel.basicPublish("", queueName, null, message.getBytes());
  53.  
  54.             System.out.println("消息发送成功!!!");
  55.         } catch (Exception ex) {
  56.             ex.printStackTrace();
  57.         } finally {
  58.             // 7: 关闭通道
  59.             if (channel != null && channel.isOpen()) {
  60.                 try {
  61.                     channel.close();
  62.                 } catch (Exception ex) {
  63.                     ex.printStackTrace();
  64.                 }
  65.             }
  66.             // 8: 关闭连接
  67.  
  68.             if (connection != null && connection.isOpen()) {
  69.                 try {
  70.                     connection.close();
  71.                 } catch (Exception ex) {
  72.                     ex.printStackTrace();
  73.                 }
  74.             }
  75.         }
  76.  
  77.  
  78.     }
  79. }

 

 

消费者

  1. package com.chen.rabbitmq.simple;
  2.  
  3. import com.rabbitmq.client.*;
  4. import java.io.IOException;
  5.  
  6. public class Consumer {
  7.  
  8.  
  9.     public static void main(String[] args) {
  10.  
  11.         // 所有的中间件技术都是基于tcp/ip协议基础之上构建新型的协议规范，只不过rabbitmq遵循的是amqp
  12.         // ip port
  13.  
  14.         // 1: 创建连接工程
  15.         ConnectionFactory connectionFactory = new ConnectionFactory();
  16.         connectionFactory.setHost("128.197.157.151");//服务器IP
  17.         connectionFactory.setPort(5672);
  18.         connectionFactory.setUsername("chenjinxian");
  19.         connectionFactory.setPassword("chenjinxian");
  20.         connectionFactory.setVirtualHost("/");
  21.  
  22.         Connection connection = null;
  23.         Channel channel = null;
  24.         try {
  25.             // 2: 创建连接Connection
  26.             connection = connectionFactory.newConnection("消费者");
  27.             // 3: 通过连接获取通道Channel
  28.             channel = connection.createChannel();
  29.             // 4: 通过通创建交换机，声明队列，绑定关系，路由key,发送消息，和接收消息
  30.  
  31.  
  32.             // true = ack 正常的逻辑是没问题 死循环 rabbit 重发策略
  33.             // false = nack 消息这在消费消息的时候可能会异常和故障
  34.             final  Channel channel2 = channel;
  35.             channel2.basicConsume("queue1", false, new DeliverCallback() {
  36.                 public void handle(String consumerTag, Delivery message) throws IOException {
  37.                     try {
  38.                         System.out.println("收到消息是" + new String(message.getBody(), "UTF-8"));
  39.                         channel2.basicAck(message.getEnvelope().getDeliveryTag(),false);
  40.                     }catch (Exception ex){
  41.                         ex.printStackTrace();
  42.                         // 三次确认 -- reject + sixin
  43.                     }
  44.  
  45.                 }
  46.             }, new CancelCallback() {
  47.                 public void handle(String consumerTag) throws IOException {
  48.                     System.out.println("接受失败了...");
  49.                 }
  50.             });
  51.  
  52.             System.out.println("开始接受消息");
  53.             System.in.read();
  54.  
  55.         } catch (Exception ex) {
  56.             ex.printStackTrace();
  57.         } finally {
  58.             // 7: 关闭通道
  59.             if (channel != null && channel.isOpen()) {
  60.                 try {
  61.                     channel.close();
  62.                 } catch (Exception ex) {
  63.                     ex.printStackTrace();
  64.                 }
  65.             }
  66.             // 8: 关闭连接
  67.  
  68.             if (connection != null && connection.isOpen()) {
  69.                 try {
  70.                     connection.close();
  71.                 } catch (Exception ex) {
  72.                     ex.printStackTrace();
  73.                 }
  74.             }
  75.         }
  76.  
  77.  
  78.     }
  79. }

 

2. 什么是AMQP

01 什么是AMQP

AMQP全称：Advanced Message Queuing Protocol（高级消息队列协议）。是应用层协议的一个开发标准，为面向消息的中间件设计

02 AMQP生产者流转过程

 

03 AMQP消费者流转过程

 

3. RabbitMQ的核心组成部分

01 RabbitMQ的核心组成部分

 

核心概念: 核心概念：
Server :又称Broker ,接受客户端的连接，实现AMQP实体服务。安装rabbitmq-serverConnection:连接，应用程序与Broker的网络连接TCP/IP/三次握手和四次挥手 服务器：又称Broker，接受客户端的连接，实现AMQP实体服务。安装Rabbitmq-serverConnection：连接，应用程序与Broker的网络连接tcp/ip/三次握手和四次挥手
Channel:网络信道，几乎所有的操作都在Channel中进行，Channel是进 息读写的通道，客户端可以建立对恪Channel，每个Channel代表一个会话任务。 频道：网络信道，几乎所有的操作都在频道中进行频道，是进息读写的通道，客户端可以建立对恪频道频道，每个频道代表一个会话任务频道。
Message :消息:服务与应用程序之间传送的数据，由Properties和body组成，Properties可是对消息进行修饰,如消息的优先级，延迟等高级特性，Body则就是消息体的内容。 消息：消息:服务与应用程序之间传送的数据，由Properties和Body组成，Properties可是对消息进行修饰，如消息的优先级，延迟等高级特性，Body则就是消息体的内容。
Virtual Host虚拟地址，用于进行逻辑隔离，最上层的消息路由，一个虚拟主机理由可以有若干个Exhange和Queueu，同一个虚拟主机里面不能有相同名字的Exchange 虚拟主机虚拟地址，用于进行逻辑隔离，最上层的消息路由，一个虚拟主机理由可以有若干个Exhange和Queueu，同一个虚拟主机里面不能有相同名字的Exchange
Exchange :交换机，接受消息，根据路由键发送消息到绑定的队列。(==不具备消息存储的能力==)Bindings : Exchange和Queue之间的虚拟连接，binding中可以保护多个routing key. 交换：交换机，接受消息，根据路由键发送消息到绑定的队列.(=不具备消息存储的能力==)绑定：Exchange和Queue之间的虚拟连接，Binding中可以保护多个路由密钥。
Routing key :是一个路由规则，虚拟机可以用它来确定如何路由一个特疋消恳.bttos:/bloq.csdn.net/qg _4485823(Queue:队列:也成为Message Queue,消息队列，保存消息并将它们转发给消费苦。"gwa" 路由密钥：是一个路由规则，虚拟机可以用它来确定如何路由一个特征消息(队列：队列:也成为消息队列，消息队列，保存消息并将它们转发给消费者.

 

02 RabbitMQ整体架构是什么样子的？

03 RabbitMQ的运行流程

 

 

4. RabbitMQ入门案例 - fanout 模式

01 RabbitMQ的模式之发布订阅模式

 

发布订阅模式的具体实现

  1. 类型：fanout

  2. 特点：Fanout - 发布与订阅模式，是一种广播机制，它是没有路由 key的模式

（注意这里已经在可视化界面让队列绑定了交换机）

 

生产者

  1. package com.chen.rabbitmq.fanout;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6.  
  7. /**
  8.  发布订阅模式的具体实现
  9.  类型：fanout
  10.  特点：Fanout - 发布与订阅模式，是一种广播机制，它是没有路由 key的模式
  11.  */
  12. public class Producer {
  13.     public static void main(String[] args) {
  14.         // 1: 创建连接工厂
  15.         ConnectionFactory connectionFactory = new ConnectionFactory();
  16.         // 2: 设置连接属性
  17.         connectionFactory.setHost("128.156.157.161");
  18.         connectionFactory.setPort(5672);
  19.         connectionFactory.setVirtualHost("/");
  20.         connectionFactory.setUsername("chenjinxian");
  21.         connectionFactory.setPassword("chenjinxian");
  22.         Connection connection = null;
  23.         Channel channel = null;
  24.         try {
  25.             // 3: 从连接工厂中获取连接
  26.             connection = connectionFactory.newConnection("生产者");
  27.             // 4: 从连接中获取通道channel
  28.             channel = connection.createChannel();
  29.  
  30.             // 5： 准备发送消息的内容
  31.             String message = "hello xuexi!!!";
  32.  
  33.             // 6：准备交换机
  34.             String exchangeName = "fanout_change";
  35.  
  36.             // 8: 指定交换机的类型
  37.             String type = "fanout";
  38.             // 7: 发送消息给中间件rabbitmq-server
  39.             // @params1: 交换机exchange
  40.             // @params2: 队列名称/routingkey
  41.             // @params3: 属性配置
  42.             // @params4: 发送消息的内容
  43.             //  #.course.* queue3
  44.             // *.order.# queue2 ta
  45.             // com.order.course.xxx collecion
  46.             channel.basicPublish(exchangeName,"", null, message.getBytes());
  47.  
  48.  
  49.             System.out.println("消息发送成功!");
  50.         } catch (Exception ex) {
  51.             ex.printStackTrace();
  52.             System.out.println("发送消息出现异常...");
  53.         } finally {
  54.  
  55.             // 7: 释放连接关闭通道
  56.             if (channel != null && channel.isOpen()) {
  57.                 try {
  58.                     channel.close();
  59.                 } catch (Exception ex) {
  60.                     ex.printStackTrace();
  61.                 }
  62.             }
  63.             if (connection != null) {
  64.                 try {
  65.                     connection.close();
  66.                 } catch (Exception ex) {
  67.                     ex.printStackTrace();
  68.                 }
  69.             }
  70.         }
  71.     }
  72. }

消费者

  1. package com.chen.rabbitmq.fanout;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6.  
  7. /**
  8.  发布订阅模式的具体实现
  9.  类型：fanout
  10.  特点：Fanout - 发布与订阅模式，是一种广播机制，它是没有路由 key的模式
  11.  */
  12. public class Consumer {
  13.  
  14.     private static Runnable runnable = new Runnable() {
  15.         public void run() {
  16.             // 1: 创建连接工厂
  17.             ConnectionFactory connectionFactory = new ConnectionFactory();
  18.             // 2: 设置连接属性
  19.             connectionFactory.setHost("128.156.157.151");
  20.             connectionFactory.setPort(5672);
  21.             connectionFactory.setVirtualHost("/");
  22.             connectionFactory.setUsername("chenjinxian");
  23.             connectionFactory.setPassword("chenjinxian");
  24.             //获取队列的名称
  25.             final String queueName = Thread.currentThread().getName();
  26.             Connection connection = null;
  27.             Channel channel = null;
  28.             try {
  29.                 // 3: 从连接工厂中获取连接
  30.                 connection = connectionFactory.newConnection("生产者");
  31.                 // 4: 从连接中获取通道channel
  32.                 channel = connection.createChannel();
  33.                 // 5: 申明队列queue存储消息
  34.                 /*
  35.                  *  如果队列不存在，则会创建
  36.                  *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
  37.                  *
  38.                  *  @params1： queue 队列的名称
  39.                  *  @params2： durable 队列是否持久化
  40.                  *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
  41.                  *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
  42.                  *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
  43.                  * */
  44.                 // 这里如果queue已经被创建过一次了，可以不需要定义
  45.                 //channel.queueDeclare("queue1", false, false, false, null);
  46.                 // 6： 定义接受消息的回调
  47.                 Channel finalChannel = channel;
  48.                 finalChannel.basicConsume(queueName, true, new DeliverCallback() {
  49.                     @Override
  50.                     public void handle(String s, Delivery delivery) throws IOException {
  51.                         System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));
  52.                     }
  53.                 }, new CancelCallback() {
  54.                     @Override
  55.                     public void handle(String s) throws IOException {
  56.                     }
  57.                 });
  58.                 System.out.println(queueName + "：开始接受消息");
  59.                 System.in.read();
  60.             } catch (Exception ex) {
  61.                 ex.printStackTrace();
  62.                 System.out.println("发送消息出现异常...");
  63.             } finally {
  64.                 // 7: 释放连接关闭通道
  65.                 if (channel != null && channel.isOpen()) {
  66.                     try {
  67.                         channel.close();
  68.                     } catch (Exception ex) {
  69.                         ex.printStackTrace();
  70.                     }
  71.                 }
  72.                 if (connection != null && connection.isOpen()) {
  73.                     try {
  74.                         connection.close();
  75.                     } catch (Exception ex) {
  76.                         ex.printStackTrace();
  77.                     }
  78.                 }
  79.             }
  80.         }
  81.     };
  82.  
  83.  
  84.  
  85.     public static void main(String[] args) {
  86.         // 启动三个线程去执行
  87.         new Thread(runnable, "queue1").start();
  88.         new Thread(runnable, "queue2").start();
  89.         new Thread(runnable, "queue3").start();
  90.         new Thread(runnable, "queue4").start();
  91.         //new Thread(runnable, "queue5").start();
  92.     }
  93. }

 

 

5. RabbitMQ入门案例 - Direct 模式

 （注意这里已经在可视化界面让队列绑定了交换机）

生产者

  1. package com.chen.rabbitmq.routing;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6.  
  7. /**
  8.  Direct 模式
  9.  */
  10. public class Producer {
  11.     public static void main(String[] args) {
  12.         // 1: 创建连接工厂
  13.         ConnectionFactory connectionFactory = new ConnectionFactory();
  14.         // 2: 设置连接属性
  15.         connectionFactory.setHost("128.176.157.151");
  16.         connectionFactory.setPort(5672);
  17.         connectionFactory.setVirtualHost("/");
  18.         connectionFactory.setUsername("chenjinxian");
  19.         connectionFactory.setPassword("chenjinxian");
  20.         Connection connection = null;
  21.         Channel channel = null;
  22.         try {
  23.             // 3: 从连接工厂中获取连接
  24.             connection = connectionFactory.newConnection("生产者");
  25.             // 4: 从连接中获取通道channel
  26.             channel = connection.createChannel();
  27.  
  28.             // 5： 准备发送消息的内容
  29.             String message = "hello direct_exchange!!!";
  30.  
  31.             // 6：准备交换机
  32.             String exchangeName = "direct_exchange";
  33.             // 7: 定义路由key
  34.             String routeKey = "email";
  35.             // 8: 指定交换机的类型
  36.             String type = "direct";
  37.             // 7: 发送消息给中间件rabbitmq-server
  38.             // @params1: 交换机exchange
  39.             // @params2: 队列名称/routingkey
  40.             // @params3: 属性配置
  41.             // @params4: 发送消息的内容
  42.             //  #.course.* queue3
  43.             // *.order.# queue2 ta
  44.             // com.order.course.xxx collecion
  45.             channel.basicPublish(exchangeName, routeKey, null, message.getBytes());
  46.             System.out.println("消息发送成功!");
  47.         } catch (Exception ex) {
  48.             ex.printStackTrace();
  49.             System.out.println("发送消息出现异常...");
  50.         } finally {
  51.             // 7: 释放连接关闭通道
  52.             if (channel != null && channel.isOpen()) {
  53.                 try {
  54.                     channel.close();
  55.                 } catch (Exception ex) {
  56.                     ex.printStackTrace();
  57.                 }
  58.             }
  59.             if (connection != null) {
  60.                 try {
  61.                     connection.close();
  62.                 } catch (Exception ex) {
  63.                     ex.printStackTrace();
  64.                 }
  65.             }
  66.         }
  67.     }
  68. }

消费者

  1. package com.chen.rabbitmq.routing;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6.  
  7. /**
  8.  Direct 模式
  9.  */
  10. public class Consumer {
  11.  
  12.     private static Runnable runnable = new Runnable() {
  13.         public void run() {
  14.             // 1: 创建连接工厂
  15.             ConnectionFactory connectionFactory = new ConnectionFactory();
  16.             // 2: 设置连接属性
  17.             connectionFactory.setHost("123.156.147.151");
  18.             connectionFactory.setPort(5672);
  19.             connectionFactory.setVirtualHost("/");
  20.             connectionFactory.setUsername("chenjinxian");
  21.             connectionFactory.setPassword("chenjinxian");
  22.             //获取队列的名称
  23.             final String queueName = Thread.currentThread().getName();
  24.             Connection connection = null;
  25.             Channel channel = null;
  26.             try {
  27.                 // 3: 从连接工厂中获取连接
  28.                 connection = connectionFactory.newConnection("生产者");
  29.                 // 4: 从连接中获取通道channel
  30.                 channel = connection.createChannel();
  31.                 // 5: 申明队列queue存储消息
  32.                 /*
  33.                  *  如果队列不存在，则会创建
  34.                  *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
  35.                  *
  36.                  *  @params1： queue 队列的名称
  37.                  *  @params2： durable 队列是否持久化
  38.                  *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
  39.                  *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
  40.                  *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
  41.                  * */
  42.                 // 这里如果queue已经被创建过一次了，可以不需要定义
  43.                 //channel.queueDeclare("queue1", false, false, false, null);
  44.                 // 6： 定义接受消息的回调
  45.                 Channel finalChannel = channel;
  46.                 finalChannel.basicConsume(queueName, true, new DeliverCallback() {
  47.                     @Override
  48.                     public void handle(String s, Delivery delivery) throws IOException {
  49.                         System.out.println(queueName + "：收到消息是：" + new String(delivery.getBody(), "UTF-8"));
  50.                     }
  51.                 }, new CancelCallback() {
  52.                     @Override
  53.                     public void handle(String s) throws IOException {
  54.                     }
  55.                 });
  56.                 System.out.println(queueName + "：开始接受消息");
  57.                 System.in.read();
  58.             } catch (Exception ex) {
  59.                 ex.printStackTrace();
  60.                 System.out.println("发送消息出现异常...");
  61.             } finally {
  62.                 // 7: 释放连接关闭通道
  63.                 if (channel != null && channel.isOpen()) {
  64.                     try {
  65.                         channel.close();
  66.                     } catch (Exception ex) {
  67.                         ex.printStackTrace();
  68.                     }
  69.                 }
  70.                 if (connection != null && connection.isOpen()) {
  71.                     try {
  72.                         connection.close();
  73.                     } catch (Exception ex) {
  74.                         ex.printStackTrace();
  75.                     }
  76.                 }
  77.             }
  78.         }
  79.     };
  80.  
  81.  
  82.  
  83.     public static void main(String[] args) {
  84.         // 启动三个线程去执行
  85.         new Thread(runnable, "queue1").start();
  86.         new Thread(runnable, "queue2").start();
  87.         new Thread(runnable, "queue3").start();
  88.         new Thread(runnable, "queue4").start();
  89.       //  new Thread(runnable, "queue5").start();
  90.     }
  91. }

6. RabbitMQ入门案例 - Topic 模式

  （注意这里已经在可视化界面让队列绑定了交换机）

 生产者

  1. package com.chen.rabbitmq.topics;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6.  
  7. /**
  8.  Topic模式
  9.  */
  10. public class Producer {
  11.     public static void main(String[] args) {
  12.         // 1: 创建连接工厂
  13.         ConnectionFactory connectionFactory = new ConnectionFactory();
  14.         // 2: 设置连接属性
  15.         connectionFactory.setHost("125.156.157.151");
  16.         connectionFactory.setPort(5672);
  17.         connectionFactory.setVirtualHost("/");
  18.         connectionFactory.setUsername("chenjinxian");
  19.         connectionFactory.setPassword("chenjinxian");
  20.         Connection connection = null;
  21.         Channel channel = null;
  22.         try {
  23.             // 3: 从连接工厂中获取连接
  24.             connection = connectionFactory.newConnection("生产者");
  25.             // 4: 从连接中获取通道channel
  26.             channel = connection.createChannel();
  27.  
  28.             // 5： 准备发送消息的内容
  29.             String message = "hello topic_exchange!!!";
  30.  
  31.             // 6：准备交换机
  32.             String exchangeName = "topic_exchange";
  33.             // 7: 定义路由key
  34.             String routeKey = "com.order.user";
  35.             // 8: 指定交换机的类型
  36.             String type = "topic";
  37.             // 7: 发送消息给中间件rabbitmq-server
  38.             // @params1: 交换机exchange
  39.             // @params2: 队列名称/routingkey
  40.             // @params3: 属性配置
  41.             // @params4: 发送消息的内容
  42.             //  #.course.* queue3
  43.             // *.order.# queue2 ta
  44.             // com.order.course.xxx collecion
  45.             channel.basicPublish(exchangeName, routeKey, null, message.getBytes());
  46.             System.out.println("消息发送成功!");
  47.         } catch (Exception ex) {
  48.             ex.printStackTrace();
  49.             System.out.println("发送消息出现异常...");
  50.         } finally {
  51.             // 7: 释放连接关闭通道
  52.             if (channel != null && channel.isOpen()) {
  53.                 try {
  54.                     channel.close();
  55.                 } catch (Exception ex) {
  56.                     ex.printStackTrace();
  57.                 }
  58.             }
  59.             if (connection != null) {
  60.                 try {
  61.                     connection.close();
  62.                 } catch (Exception ex) {
  63.                     ex.printStackTrace();
  64.                 }
  65.             }
  66.         }
  67.     }
  68. }

 消费者不变

完整案例（创建交换机，创建队列，交换机与队列绑定）

  1. package com.chen.rabbitmq.all;
  2.  
  3. import com.rabbitmq.client.Channel;
  4. import com.rabbitmq.client.Connection;
  5. import com.rabbitmq.client.ConnectionFactory;
  6.  
  7. /**
  8. 完整案例
  9.  */
  10. public class Producer {
  11.     public static void main(String[] args) {
  12.         // 1: 创建连接工厂
  13.         ConnectionFactory connectionFactory = new ConnectionFactory();
  14.         // 2: 设置连接属性
  15.         connectionFactory.setHost("151.156.157.151");
  16.         connectionFactory.setPort(5672);
  17.         connectionFactory.setVirtualHost("/");
  18.         connectionFactory.setUsername("chenjinxian");
  19.         connectionFactory.setPassword("chenjinxian");
  20.         Connection connection = null;
  21.         Channel channel = null;
  22.         try {
  23.             // 3: 从连接工厂中获取连接
  24.             connection = connectionFactory.newConnection("生产者");
  25.             // 4: 从连接中获取通道channel
  26.             channel = connection.createChannel();
  27.             // 6： 准备发送消息的内容
  28.             String message = " 你好，小白";
  29.             // 交换机
  30.             String  exchangeName = "direct_message_exchange";
  31.             // 交换机的类型 direct/topic/fanout/headers
  32.             String exchangeType = "direct";
  33.  
  34.             // 如果你用界面把queueu 和 exchange的关系先绑定话，你代码就不需要在编写这些声明代码可以让代码变得更加简洁，但是不容读懂
  35.             // 如果用代码的方式去声明，我们要学习一下
  36.             // 7： 声明交换机 所谓的持久化就是指，交换机会不会随着服务器重启造成丢失，如果是true代表不丢失，false重启就会丢失
  37.             channel.exchangeDeclare(exchangeName,exchangeType,true);
  38.  
  39.             // 8： 声明队列
  40.             channel.queueDeclare("queue5",true,false,false,null);
  41.             channel.queueDeclare("queue6",true,false,false,null);
  42.             channel.queueDeclare("queue7",true,false,false,null);
  43.  
  44.             // 9：绑定队列和交换机的关系
  45.             channel.queueBind("queue5",exchangeName,"order");
  46.             channel.queueBind("queue6",exchangeName,"order");
  47.             channel.queueBind("queue7",exchangeName,"course");
  48.  
  49.             channel.basicPublish(exchangeName, "course", null, message.getBytes());
  50.             System.out.println("消息发送成功!");
  51.         } catch (Exception ex) {
  52.             ex.printStackTrace();
  53.             System.out.println("发送消息出现异常...");
  54.         } finally {
  55.             // 7: 释放连接关闭通道
  56.             if (channel != null && channel.isOpen()) {
  57.                 try {
  58.                     channel.close();
  59.                 } catch (Exception ex) {
  60.                     ex.printStackTrace();
  61.                 }
  62.             }
  63.             if (connection != null) {
  64.                 try {
  65.                     connection.close();
  66.                 } catch (Exception ex) {
  67.                     ex.printStackTrace();
  68.                 }
  69.             }
  70.         }
  71.     }
  72. }

 执行完后生成队列和交换机

 

 

7. RabbitMQ入门案例 - Work模式

01 Work模式轮询模式（Round-Robin）

 

当有多个消费者时，我们的消息会被哪个消费者消费呢，我们又该如何均衡消费者消费信息的多少呢？

主要有两种模式：

  1. 轮询模式的分发：一个消费者一条，按均分配

  2. 公平分发：根据消费者的消费能力进行公平分发，处理快的处理的多，处理慢的处理的少；按劳分配

01轮询模式

生产者

  1. package com.chen.rabbitmq.work.lunxun;
  2. import com.rabbitmq.client.Channel;
  3. import com.rabbitmq.client.Connection;
  4. import com.rabbitmq.client.ConnectionFactory;
  5. /**
  6.  轮询模式
  7.  */
  8. public class Producer {
  9.     public static void main(String[] args) {
  10.         // 1: 创建连接工厂
  11.         ConnectionFactory connectionFactory = new ConnectionFactory();
  12.         // 2: 设置连接属性
  13.         connectionFactory.setHost("123.156.147.151");
  14.         connectionFactory.setPort(5672);
  15.         connectionFactory.setVirtualHost("/");
  16.         connectionFactory.setUsername("chenjinxian");
  17.         connectionFactory.setPassword("chenjinxian");
  18.         Connection connection = null;
  19.         Channel channel = null;
  20.         try {
  21.             // 3: 从连接工厂中获取连接
  22.             connection = connectionFactory.newConnection("生产者");
  23.             // 4: 从连接中获取通道channel
  24.             channel = connection.createChannel();
  25.             // 6： 准备发送消息的内容
  26.             //===============================end topic模式==================================
  27.             for (int i = 1; i <= 20; i++) {
  28.                 //消息的内容
  29.                 String msg = "学相伴:" + i;
  30.                 // 7: 发送消息给中间件rabbitmq-server
  31.                 // @params1: 交换机exchange
  32.                 // @params2: 队列名称/routingkey
  33.                 // @params3: 属性配置
  34.                 // @params4: 发送消息的内容
  35.                 channel.basicPublish("", "queue1", null, msg.getBytes());
  36.             }
  37.             System.out.println("消息发送成功!");
  38.         } catch (Exception ex) {
  39.             ex.printStackTrace();
  40.             System.out.println("发送消息出现异常...");
  41.         } finally {
  42.             // 7: 释放连接关闭通道
  43.             if (channel != null && channel.isOpen()) {
  44.                 try {
  45.                     channel.close();
  46.                 } catch (Exception ex) {
  47.                     ex.printStackTrace();
  48.                 }
  49.             }
  50.             if (connection != null) {
  51.                 try {
  52.                     connection.close();
  53.                 } catch (Exception ex) {
  54.                     ex.printStackTrace();
  55.                 }
  56.             }
  57.         }
  58.     }
  59. }

消费者

  1. package com.chen.rabbitmq.work.lunxun;
  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  4. /**
  5.  轮询模式
  6.  */
  7. public class Work1 {
  8.     public static void main(String[] args) {
  9.         // 1: 创建连接工厂
  10.         ConnectionFactory connectionFactory = new ConnectionFactory();
  11.         // 2: 设置连接属性
  12.         connectionFactory.setHost("123.156.147.155");
  13.         connectionFactory.setPort(5672);
  14.         connectionFactory.setVirtualHost("/");
  15.         connectionFactory.setUsername("chenjinxian");
  16.         connectionFactory.setPassword("chenjinxian");
  17.         Connection connection = null;
  18.         Channel channel = null;
  19.         try {
  20.             // 3: 从连接工厂中获取连接
  21.             connection = connectionFactory.newConnection("消费者-Work1");
  22.             // 4: 从连接中获取通道channel
  23.             channel = connection.createChannel();
  24.             // 5: 申明队列queue存储消息
  25.             /*
  26.              *  如果队列不存在，则会创建
  27.              *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
  28.              *
  29.              *  @params1： queue 队列的名称
  30.              *  @params2： durable 队列是否持久化
  31.              *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
  32.              *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
  33.              *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
  34.              * */
  35.             // 这里如果queue已经被创建过一次了，可以不需要定义
  36. //            channel.queueDeclare("queue1", false, false, false, null);
  37.             // 同一时刻，服务器只会推送一条消息给消费者
  38.             // 6： 定义接受消息的回调
  39.             Channel finalChannel = channel;
  40.             //finalChannel.basicQos(1);
  41.  
  42.             finalChannel.basicConsume("queue1", true, new DeliverCallback() {
  43.                 @Override
  44.                 public void handle(String s, Delivery delivery) throws IOException {
  45.                     try{
  46.                         System.out.println("Work1-收到消息是：" + new String(delivery.getBody(), "UTF-8"));
  47.                         Thread.sleep(200);
  48.                     }catch(Exception ex){
  49.                         ex.printStackTrace();
  50.                     }
  51.                 }
  52.             }, new CancelCallback() {
  53.                 @Override
  54.                 public void handle(String s) throws IOException {
  55.                 }
  56.             });
  57.             System.out.println("Work1-开始接受消息");
  58.             System.in.read();
  59.         } catch (Exception ex) {
  60.             ex.printStackTrace();
  61.             System.out.println("发送消息出现异常...");
  62.         } finally {
  63.             // 7: 释放连接关闭通道
  64.             if (channel != null && channel.isOpen()) {
  65.                 try {
  66.                     channel.close();
  67.                 } catch (Exception ex) {
  68.                     ex.printStackTrace();
  69.                 }
  70.             }
  71.             if (connection != null && connection.isOpen()) {
  72.                 try {
  73.                     connection.close();
  74.                 } catch (Exception ex) {
  75.                     ex.printStackTrace();
  76.                 }
  77.             }
  78.         }
  79.     }
  80. }
  1. package com.chen.rabbitmq.work.lunxun;
  2. import com.rabbitmq.client.*;
  3. import java.io.IOException;
  4. /**
  5.  轮询模式
  6.  */
  7. public class Work2 {
  8.     public static void main(String[] args) {
  9.         // 1: 创建连接工厂
  10.         ConnectionFactory connectionFactory = new ConnectionFactory();
  11.         // 2: 设置连接属性
  12.         connectionFactory.setHost("123.195.157.151");
  13.         connectionFactory.setPort(5672);
  14.         connectionFactory.setVirtualHost("/");
  15.         connectionFactory.setUsername("chenjinxian");
  16.         connectionFactory.setPassword("chenjinxian");
  17.         Connection connection = null;
  18.         Channel channel = null;
  19.         try {
  20.             // 3: 从连接工厂中获取连接
  21.             connection = connectionFactory.newConnection("消费者-Work2");
  22.             // 4: 从连接中获取通道channel
  23.             channel = connection.createChannel();
  24.             // 5: 申明队列queue存储消息
  25.             /*
  26.              *  如果队列不存在，则会创建
  27.              *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
  28.              *
  29.              *  @params1： queue 队列的名称
  30.              *  @params2： durable 队列是否持久化
  31.              *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
  32.              *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
  33.              *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
  34.              * */
  35.             // 这里如果queue已经被创建过一次了，可以不需要定义
  36.             //channel.queueDeclare("queue1", false, true, false, null);
  37.             // 同一时刻，服务器只会推送一条消息给消费者
  38.             //channel.basicQos(1);
  39.             // 6： 定义接受消息的回调
  40.             Channel finalChannel = channel;
  41.             //finalChannel.basicQos(1);
  42.             finalChannel.basicConsume("queue1", true, new DeliverCallback() {
  43.                 @Override
  44.                 public void handle(String s, Delivery delivery) throws IOException {
  45.                     try{
  46.                         System.out.println("Work2-收到消息是：" + new String(delivery.getBody(), "UTF-8"));
  47.                         Thread.sleep(100);
  48.                     }catch(Exception ex){
  49.                         ex.printStackTrace();
  50.                     }
  51.                 }
  52.             }, new CancelCallback() {
  53.                 @Override
  54.                 public void handle(String s) throws IOException {
  55.                 }
  56.             });
  57.             System.out.println("Work2-开始接受消息");
  58.             System.in.read();
  59.         } catch (Exception ex) {
  60.             ex.printStackTrace();
  61.             System.out.println("发送消息出现异常...");
  62.         } finally {
  63.             // 7: 释放连接关闭通道
  64.             if (channel != null && channel.isOpen()) {
  65.                 try {
  66.                     channel.close();
  67.                 } catch (Exception ex) {
  68.                     ex.printStackTrace();
  69.                 }
  70.             }
  71.             if (connection != null && connection.isOpen()) {
  72.                 try {
  73.                     connection.close();
  74.                 } catch (Exception ex) {
  75.                     ex.printStackTrace();
  76.                 }
  77.             }
  78.         }
  79.     }
  80. }

 

02 Work模式公平分发模式 

生产者

  1. package com.chen.rabbitmq.work.fair;
  2. import com.rabbitmq.client.Channel;
  3. import com.rabbitmq.client.Connection;
  4. import com.rabbitmq.client.ConnectionFactory;
  5.  
  6. /**
  7.  公平分发模式
  8.  */
  9. public class Producer {
  10.     public static void main(String[] args) {
  11.         // 1: 创建连接工厂
  12.         ConnectionFactory connectionFactory = new ConnectionFactory();
  13.         // 2: 设置连接属性
  14.         connectionFactory.setHost("125.156.157.151");
  15.         connectionFactory.setPort(5672);
  16.         connectionFactory.setVirtualHost("/");
  17.         connectionFactory.setUsername("chenjinxian");
  18.         connectionFactory.setPassword("chenjinxian");
  19.         Connection connection = null;
  20.         Channel channel = null;
  21.         try {
  22.             // 3: 从连接工厂中获取连接
  23.             connection = connectionFactory.newConnection("生产者");
  24.             // 4: 从连接中获取通道channel
  25.             channel = connection.createChannel();
  26.             // 6： 准备发送消息的内容
  27.             //===============================end topic模式==================================
  28.             for (int i = 1; i <= 20; i++) {
  29.                 //消息的内容
  30.                 String msg = "学相伴:" + i;
  31.                 // 7: 发送消息给中间件rabbitmq-server
  32.                 // @params1: 交换机exchange
  33.                 // @params2: 队列名称/routingkey
  34.                 // @params3: 属性配置
  35.                 // @params4: 发送消息的内容
  36.                 channel.basicPublish("", "queue1", null, msg.getBytes());
  37.             }
  38.             System.out.println("消息发送成功!");
  39.         } catch (Exception ex) {
  40.             ex.printStackTrace();
  41.             System.out.println("发送消息出现异常...");
  42.         } finally {
  43.             // 7: 释放连接关闭通道
  44.             if (channel != null && channel.isOpen()) {
  45.                 try {
  46.                     channel.close();
  47.                 } catch (Exception ex) {
  48.                     ex.printStackTrace();
  49.                 }
  50.             }
  51.             if (connection != null) {
  52.                 try {
  53.                     connection.close();
  54.                 } catch (Exception ex) {
  55.                     ex.printStackTrace();
  56.                 }
  57.             }
  58.         }
  59.     }
  60. }

消费者

  1. package com.chen.rabbitmq.work.fair;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6.  
  7. /**
  8.  公平分发模式
  9.  */
  10. public class Work1 {
  11.     public static void main(String[] args) {
  12.         // 1: 创建连接工厂
  13.         ConnectionFactory connectionFactory = new ConnectionFactory();
  14.         // 2: 设置连接属性
  15.         connectionFactory.setHost("123.156.146.151");
  16.         connectionFactory.setPort(5672);
  17.         connectionFactory.setVirtualHost("/");
  18.         connectionFactory.setUsername("chenjinxian");
  19.         connectionFactory.setPassword("chenjinxian");
  20.         Connection connection = null;
  21.         Channel channel = null;
  22.         try {
  23.             // 3: 从连接工厂中获取连接
  24.             connection = connectionFactory.newConnection("消费者-Work1");
  25.             // 4: 从连接中获取通道channel
  26.             channel = connection.createChannel();
  27.             // 5: 申明队列queue存储消息
  28.             /*
  29.              *  如果队列不存在，则会创建
  30.              *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
  31.              *
  32.              *  @params1： queue 队列的名称
  33.              *  @params2： durable 队列是否持久化
  34.              *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
  35.              *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
  36.              *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
  37.              * */
  38.             // 这里如果queue已经被创建过一次了，可以不需要定义
  39. //            channel.queueDeclare("queue1", false, false, false, null);
  40.             // 同一时刻，服务器只会推送一条消息给消费者
  41.             // 6： 定义接受消息的回调
  42.             final Channel finalChannel = channel;
  43.             
  44.             finalChannel.basicQos(1);
  45.             finalChannel.basicConsume("queue1", false, new DeliverCallback() {
  46.                 @Override
  47.                 public void handle(String s, Delivery delivery) throws IOException {
  48.                     try{
  49.                         System.out.println("Work1-收到消息是：" + new String(delivery.getBody(), "UTF-8"));
  50.                         Thread.sleep(1000);
  51.                         // 改成手动应答
  52.                         finalChannel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
  53.                     }catch(Exception ex){
  54.                         ex.printStackTrace();
  55.                     }
  56.                 }
  57.             }, new CancelCallback() {
  58.                 @Override
  59.                 public void handle(String s) throws IOException {
  60.                 }
  61.             });
  62.             System.out.println("Work1-开始接受消息");
  63.             System.in.read();
  64.         } catch (Exception ex) {
  65.             ex.printStackTrace();
  66.             System.out.println("发送消息出现异常...");
  67.         } finally {
  68.             // 7: 释放连接关闭通道
  69.             if (channel != null && channel.isOpen()) {
  70.                 try {
  71.                     channel.close();
  72.                 } catch (Exception ex) {
  73.                     ex.printStackTrace();
  74.                 }
  75.             }
  76.             if (connection != null && connection.isOpen()) {
  77.                 try {
  78.                     connection.close();
  79.                 } catch (Exception ex) {
  80.                     ex.printStackTrace();
  81.                 }
  82.             }
  83.         }
  84.     }
  85. }
  1. package com.chen.rabbitmq.work.fair;
  2.  
  3. import com.rabbitmq.client.*;
  4.  
  5. import java.io.IOException;
  6.  
  7. /**
  8.  公平分发模式
  9.  */
  10. public class Work2 {
  11.     public static void main(String[] args) {
  12.         // 1: 创建连接工厂
  13.         ConnectionFactory connectionFactory = new ConnectionFactory();
  14.         // 2: 设置连接属性
  15.         connectionFactory.setHost("121.156.157.131");
  16.         connectionFactory.setPort(5672);
  17.         connectionFactory.setVirtualHost("/");
  18.         connectionFactory.setUsername("chenjinxian");
  19.         connectionFactory.setPassword("chenjinxian");
  20.         Connection connection = null;
  21.         Channel channel = null;
  22.         try {
  23.             // 3: 从连接工厂中获取连接
  24.             connection = connectionFactory.newConnection("消费者-Work2");
  25.             // 4: 从连接中获取通道channel
  26.             channel = connection.createChannel();
  27.             // 5: 申明队列queue存储消息
  28.             /*
  29.              *  如果队列不存在，则会创建
  30.              *  Rabbitmq不允许创建两个相同的队列名称，否则会报错。
  31.              *
  32.              *  @params1： queue 队列的名称
  33.              *  @params2： durable 队列是否持久化
  34.              *  @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，其他的通道不能访问，并且连接自动关闭
  35.              *  @params4： autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除消息。
  36.              *  @params5： arguments 可以设置队列附加参数，设置队列的有效期，消息的最大长度，队列的消息生命周期等等。
  37.              * */
  38.             // 这里如果queue已经被创建过一次了，可以不需要定义
  39.             //channel.queueDeclare("queue1", false, true, false, null);
  40.             // 同一时刻，服务器只会推送一条消息给消费者
  41.             //channel.basicQos(1);
  42.             // 6： 定义接受消息的回调
  43.             final Channel finalChannel = channel;
  44.             finalChannel.basicQos(1);
  45.             finalChannel.basicConsume("queue1", false, new DeliverCallback() {
  46.                 @Override
  47.                 public void handle(String s, Delivery delivery) throws IOException {
  48.                     try{
  49.                         System.out.println("Work2-收到消息是：" + new String(delivery.getBody(), "UTF-8"));
  50.                         Thread.sleep(200);
  51.                         // 一定使用我们的手动应答
  52.                         finalChannel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
  53.                     }catch(Exception ex){
  54.                         ex.printStackTrace();
  55.                     }
  56.                 }
  57.             }, new CancelCallback() {
  58.                 @Override
  59.                 public void handle(String s) throws IOException {
  60.                 }
  61.             });
  62.             System.out.println("Work2-开始接受消息");
  63.             System.in.read();
  64.         } catch (Exception ex) {
  65.             ex.printStackTrace();
  66.             System.out.println("发送消息出现异常...");
  67.         } finally {
  68.             // 7: 释放连接关闭通道
  69.             if (channel != null && channel.isOpen()) {
  70.                 try {
  71.                     channel.close();
  72.                 } catch (Exception ex) {
  73.                     ex.printStackTrace();
  74.                 }
  75.             }
  76.             if (connection != null && connection.isOpen()) {
  77.                 try {
  78.                     connection.close();
  79.                 } catch (Exception ex) {
  80.                     ex.printStackTrace();
  81.                 }
  82.             }
  83.         }
  84.     }
  85. }

8. RabbitMQ使用场景

 

01 解耦、削峰、异步

 

同步异步的问题（串行）

串行方式：将订单信息写入数据库成功后，发送注册邮件，再发送注册短信。以上三个任务全部完成后，返回给客户端

 

并行方式 异步线程池

并行方式：将订单信息写入数据库成功后，发送注册邮件的同时，发送注册短信。以上三个任务完成后，返回给客户端。与串行的差别是，并行的方式可以提高处理的时间

存在问题

  1. 耦合度高

  2. 需要自己写线程池自己维护成本太高

  3. 出现了消息可能会丢失，需要你自己做消息补偿

  4. 如何保证消息的可靠性你自己写

  5. 如果服务器承载不了，你需要自己去写高可用

异步消息队列的方式

 

 

好处：

  1. 完全解耦，用 MQ建立桥接

  2. 有独立的线程池和运行模型

  3. 出现了消息可能会丢失，MQ有持久化功能

  4. 如何保证消息的可靠性，死信队列和消息转移等

  5. 如果服务器承载不了，你需要自己去写高可用，HA镜像模型高可用

按照以上约定，用户的响应时间相当于是订单信息写入数据库的时间，也就是50毫秒。注册邮件，发送短信写入消息队列后，直接返回，因此写入消息队列的速度很快，基本可以忽略，因此用户的响应时间可能是50毫秒。因此架构改变后，系统的吞吐量提高到每秒20QPS。比串行提高了3倍，比并行提高了两倍

 

02 高内聚，低耦合

 

 

好处：

  1. 完全解耦，用 MQ建立桥接

  2. 有独立的线程池和运行模型

  3. 出现了消息可能会丢失，MQ有持久化功能

  4. 如何保证消息的可靠性，死信队列和消息转移等

  5. 如果服务器承载不了，你需要自己去写高可用，HA镜像模型高可用

按照以上约定，用户的响应时间相当于是订单信息写入数据库的时间，也就是50毫秒。注册邮件，发送短信写入消息队列后，直接返回，因此写入消息队列的速度很快，基本可以忽略，因此用户的响应时间可能是50毫秒。因此架构改变后，系统的吞吐量提高到每秒20QPS。比串行提高了3倍，比并行提高了两倍

四、Springboot案例

1. Fanout 模式

生产者

导入依赖

  1.   <!--rabbitmq starter 依赖-->
  2.         <dependency>
  3.             <groupId>org.springframework.boot</groupId>
  4.             <artifactId>spring-boot-starter-amqp</artifactId>
  5.         </dependency>

application.yml

  1. # 服务端口
  2. server:
  3.   port: 8080
  4. # 配置rabbitmq服务
  5. spring:
  6. 	rabbitmq:
  7. 		username: admin
  8. 		password: admin
  9. 		virtual-host: /
  10. 		host: 127.0.0.1
  11. 		port: 5672

目录结构 

 创建配置类RabbitMqConfiguration.java

  1. package com.chen.springbootorderrabbitmqproducer.config;
  2. import org.springframework.amqp.core.Binding;
  3. import org.springframework.amqp.core.BindingBuilder;
  4. import org.springframework.amqp.core.FanoutExchange;
  5. import org.springframework.amqp.core.Queue;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  8.  
  9. /**
  11.  */
  12. @Configuration
  13. public class RabbitMqConfiguration {
  14.  
  15.  
  16.     // 1： 声明交换机
  17.     @Bean
  18.     public FanoutExchange fanoutExchange() {
  19.         return new FanoutExchange("fanout_order_ex", true, false);
  20.     }
  21.  
  22.  
  23.     // 2: 声明队列 duanxin.fanout.queue
  24.     @Bean
  25.     public Queue duanxinqueue() {
  26.         return new Queue("duanxin.fanout.queue", true);
  27.     }
  28.  
  29.  
  30.     // 2: 声明队列 duanxin.fanout.queue
  31.     @Bean
  32.     public Queue smsqueue() {
  33.         return new Queue("sms.fanout.queue", true);
  34.     }
  35.  
  36.  
  37.     // 2: 声明队列 duanxin.fanout.queue
  38.     @Bean
  39.     public Queue emailqueue() {
  40.         return new Queue("email.fanout.queue", true);
  41.     }
  42.  
  43.  
  44.     // 3: 确定绑定关系
  45.     @Bean
  46.     public Binding bindduanxin(){
  47.         return BindingBuilder.bind(duanxinqueue()).to(fanoutExchange());
  48.     }
  49.  
  50.     // 3: 确定绑定关系
  51.     @Bean
  52.     public Binding bindsms(){
  53.         return BindingBuilder.bind(smsqueue()).to(fanoutExchange());
  54.     }
  55.  
  56.     // 3: 确定绑定关系
  57.     @Bean
  58.     public Binding bindemail(){
  59.         return BindingBuilder.bind(emailqueue()).to(fanoutExchange());
  60.     }
  61. }

 编写实现类OrderService.java

  1. package com.chen.springbootorderrabbitmqproducer.service;
  2.  
  3. import org.springframework.amqp.AmqpException;
  4. import org.springframework.amqp.core.Message;
  5. import org.springframework.amqp.core.MessagePostProcessor;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Service;
  9.  
  10. import java.util.UUID;
  11.  
  12. /**
  14.  */
  15. @Service
  16. public class OrderService {
  17.     
  18.     @Autowired
  19.     private RabbitTemplate rabbitTemplate;
  20.  
  21.     // 交换机
  22.     private String exchangeName = "fanout_order_ex";
  23.     // 路由key
  24.     private String routingKey = "";
  25.  
  26.     
  27.     /**
  28.      * @Author xuke
  29.      * @Description 模拟用户购买商品下单的业务
  30.      * @Date 22:26 2021/3/5
  31.      * @Param [userId, productId, num]
  32.      * @return void
  33.     **/
  34.     public void makeOrder(String userId,String productId,int num){
  35.         // 1: 根据商品id查询库存是否充足
  36.         // 2: 保存订单
  37.         String orderId = UUID.randomUUID().toString();
  38.         System.out.println("保存订单成功：id是：" + orderId);
  39.         // 3: 发送消息
  40.         rabbitTemplate.convertAndSend(exchangeName,routingKey,orderId);
  41.     }
  42.  
  43. }

编写测试类

  1. package com.chen.springbootorderrabbitmqproducer.rabbitmq.springbootorderrabbitmqproducer;
  2.  
  3.  
  4. import com.chen.springbootorderrabbitmqproducer.service.OrderService;
  5. import org.junit.jupiter.api.Test;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.boot.test.context.SpringBootTest;
  8.  
  9. @SpringBootTest
  10. class SpringbootOrderRabbitmqProducerApplicationTests {
  11.  
  12.  
  13.     @Autowired
  14.     private OrderService orderService;
  15.  
  16.  
  17.     @Test
  18.     public void contextLoads() {
  19.         orderService.makeOrder("100","100",10);
  20.     }
  21.  
  22.  
  23.  
  24.     @Test
  25.     public void testDirect() {
  26.         orderService.makeOrderDirect("100","100",10);
  27.     }
  28.  
  29.     @Test
  30.     public void testDirectTTl() {
  31.         orderService.makeOrderDirectTtl("100","100",10);
  32.     }
  33.  
  34.  
  35. //    @Test
  36. //    public void testTopic() {
  37. //        orderService.makeOrderTopic("100","100",10);
  38. //    }
  39.  
  40. }

消费者

application.yml

  1. # 服务端口
  2. server:
  3.   port: 8080
  4. # 配置rabbitmq服务
  5. spring:
  6. 	rabbitmq:
  7. 		username: admin
  8. 		password: admin
  9. 		virtual-host: /
  10. 		host: 127.0.0.1
  11. 		port: 5672

接受消息

  1. package com.chen.direct;
  2.  
  3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Service;
  6.  
  7. /**
  8.  
  9.  */
  10. @Service
  11. @RabbitListener(queues ={"duanxin.direct.queue"})
  12. public class DirectDuanxinConsumber {
  13.  
  14.  
  15.     // 告诉你的接收服务器的消息，没有返回值
  16.     @RabbitHandler
  17.     public void reviceMessage(String message){
  18.         System.out.println("duanxin--direct--->接收到订单消息，订单id是: " + message);
  19.     }
  20. }
  1. package com.chen.direct;
  2.  
  3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Service;
  6.  
  7. /**
  9.  */
  10. @Service
  11. @RabbitListener(queues ={"email.direct.queue"})
  12. public class DirectEmailConsumber {
  13.  
  14.  
  15.     @RabbitHandler
  16.     public void reviceMessage(String message){
  17.         System.out.println("email---direct-->接收到订单消息，订单id是: " + message);
  18.     }
  19. }
  1. package com.chen.direct;
  2.  
  3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Service;
  6.  
  7. /**
  9.  */
  10. @Service
  11. @RabbitListener(queues ={"sms.direct.queue"})
  12. public class DirectSmsConsumber {
  13.  
  14.  
  15.     @RabbitHandler
  16.     public void reviceMessage(String message){
  17.         System.out.println("sms--direct--->接收到订单消息，订单id是: " + message);
  18.     }
  19. }

2. Direct 模式

生产者

配置类

  1. package com.chen.springbootorderrabbitmqproducer.config;
  2.  
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.context.annotation.Bean;
  5. import org.springframework.context.annotation.Configuration;
  6.  
  7. import java.util.HashMap;
  8. import java.util.Map;
  9.  
  10. /**
  12.  */
  13. @Configuration
  14. public class RabbitMqConfiguration2 {
  15.  
  16.  
  17.     // 1： 声明交换机
  18.     @Bean
  19.     public DirectExchange directExchange() {
  20.         return new DirectExchange("direct_order_ex", true, false);
  21.     }
  22.  
  23.  
  24.     // 2: 声明队列 duanxin.direct.queue
  25.     @Bean
  26.     public Queue duanxinqueue() {
  27.         return new Queue("duanxin.direct.queue", true);
  28.     }
  29.  
  30.  
  31.     // 2: 声明队列 duanxin.direct.queue
  32.     @Bean
  33.     public Queue smsqueue() {
  34.         return new Queue("sms.direct.queue", true);
  35.     }
  36.  
  37.  
  38.     // 2: 声明队列 duanxin.direct.queue
  39.     @Bean
  40.     public Queue emailqueue() {
  41.         return new Queue("email.direct.queue", true);
  42.     }
  43.  
  44.  
  45.  
  46.     // 3: 确定绑定关系
  47.     @Bean
  48.     public Binding bindduanxin(){
  49.         return BindingBuilder.bind(duanxinqueue()).to(directExchange()).with("msg");
  50.     }
  51.  
  52.     // 3: 确定绑定关系
  53.     @Bean
  54.     public Binding bindsms(){
  55.         return BindingBuilder.bind(smsqueue()).to(directExchange()).with("sms");
  56.     }
  57.  
  58.     // 3: 确定绑定关系
  59.     @Bean
  60.     public Binding bindemail(){
  61.         return BindingBuilder.bind(emailqueue()).to(directExchange()).with("email");
  62.     }
  63. }

实现类：

  1. package com.chen.springbootorderrabbitmqproducer.service;
  2.  
  3. import org.springframework.amqp.AmqpException;
  4. import org.springframework.amqp.core.Message;
  5. import org.springframework.amqp.core.MessagePostProcessor;
  6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Service;
  9.  
  10. import java.util.UUID;
  11.  
  12. /**
  14.  */
  15. @Service
  16. public class OrderService {
  17.     
  18.     @Autowired
  19.     private RabbitTemplate rabbitTemplate;
  20.  
  21.     // 交换机
  22.     private String exchangeName = "fanout_order_ex";
  23.     // 路由key
  24.     private String routingKey = "";
  25.  
  26.     
  27.  
  28.  
  29.  
  30.     /**
  31.      * @Description 模拟用户购买商品下单的业务  
  32.      * @Param [userId, productId, num]
  33.      * @return void
  34.      **/
  35.     public void makeOrderDirect(String userId,String productId,int num){
  36.         // 1: 根据商品id查询库存是否充足
  37.         // 2: 保存订单
  38.         String orderId = UUID.randomUUID().toString();
  39.         System.out.println("保存订单成功：id是：" + orderId);
  40.         // 3: 发送消息
  41.         rabbitTemplate.convertAndSend("direct_order_ex","email",orderId);
  42.         rabbitTemplate.convertAndSend("direct_order_ex","sms",orderId);
  43.     }
  44.  
  45.  
  46.  
  47.  
  48.  
  49.  
  50. }

测试：

  1. package com.chen.springbootorderrabbitmqproducer.rabbitmq.springbootorderrabbitmqproducer;
  2.  
  3.  
  4. import com.chen.springbootorderrabbitmqproducer.service.OrderService;
  5. import org.junit.jupiter.api.Test;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.boot.test.context.SpringBootTest;
  8.  
  9. @SpringBootTest
  10. class SpringbootOrderRabbitmqProducerApplicationTests {
  11.  
  12.     @Autowired
  13.     private OrderService orderService;
  14.  
  15.  
  16.     @Test
  17.     public void testDirect() {
  18.         orderService.makeOrderDirect("100","100",10);
  19.     }
  20.  
  21.  
  22. }

消费者 

  1. package com.chen.direct;
  2.  
  3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Service;
  6.  
  7. /**
  8.  * @description:
  9.  */
  10. @Service
  11. @RabbitListener(queues ={"duanxin.direct.queue"})
  12. public class DirectDuanxinConsumber {
  13.  
  14.  
  15.     // 告诉你的接收服务器的消息，没有返回值
  16.     @RabbitHandler
  17.     public void reviceMessage(String message){
  18.         System.out.println("duanxin--direct--->接收到订单消息，订单id是: " + message);
  19.     }
  20. }
  1. package com.chen.direct;
  2.  
  3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Service;
  6.  
  7. /**
  9.  */
  10. @Service
  11. @RabbitListener(queues ={"email.direct.queue"})
  12. public class DirectEmailConsumber {
  13.  
  14.  
  15.     @RabbitHandler
  16.     public void reviceMessage(String message){
  17.         System.out.println("email---direct-->接收到订单消息，订单id是: " + message);
  18.     }
  19. }
  1. package com.chen.direct;
  2.  
  3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  5. import org.springframework.stereotype.Service;
  6.  
  7. /**
  8.  
  9. @Service
  10. @RabbitListener(queues ={"sms.direct.queue"})
  11. public class DirectSmsConsumber {
  12.  
  13.  
  14.     @RabbitHandler
  15.     public void reviceMessage(String message){
  16.         System.out.println("sms--direct--->接收到订单消息，订单id是: " + message);
  17.     }
  18. }

3. Topic 模式

生产者

  1. public class OrderService{
  2.     @Autowired
  3.     private RabbitTemplate rabbitTemplate;
  4.     //模拟用户下单
  5.     public void makeOrder(String userid,String productid,int num){
  6.   
  7.   public void makeOrderTopic(String userId,String productId,int num){
  8.         // 1: 根据商品id查询库存是否充足
  9.         // 2: 保存订单
  10.         String orderId = UUID.randomUUID().toString();
  11.         System.out.println("保存订单成功：id是：" + orderId);
  12.         // 3: 发送消息
  13.  
  14.         //com.#  duanxin
  15.         //#.email.* email
  16.         //#.sms.# sms
  17.         // 设置消息确认机制
  18.         rabbitTemplate.convertAndSend("topic_order_ex","com.email.sms.xxx",orderId);
  19.     }
  20. }
  1.  @Test
  2.     public void testTopic() {
  3.         orderService.makeOrderTopic("100","100",10);
  4.     }

 

消费者（采用注解）

FanoutSmsConsumer.java

  1. @Component
  2. @RabbitListener(bindings = @QueueBinding(
  3. 	value = @Queue(value = "sms.topic.queue",durable = "true",antoDelete = "false"),
  4.     exchange = @Exchange(value = "topic_order_exchange",type = "ExchangeTypes.TOPIC")
  5.     key = "#.sms.#"
  6. ))
  7. public class TopicSmsConsumer{
  8.     @RabbitHandler
  9.     public void reviceMessage(String message){
  10.         sout("sms接收到了的订单信息是："+message);
  11.     }
  12. }

FanoutDuanxinConsumer.java

  1. @Component
  2. @RabbitListener(bindings = @QueueBinding(
  3. 	value = @Queue(value = "duanxin.topic.queue",durable = "true",antoDelete = "false"),
  4.     exchange = @Exchange(value = "topic_order_exchange",type = "ExchangeTypes.TOPIC")
  5.     key = "#.duanxin.#"
  6. ))
  7. public classTopicDuanxinConsumer{
  8.     @RabbitHandler
  9.     public void reviceMessage(String message){
  10.         sout("duanxin接收到了的订单信息是："+message);
  11.     }
  12. }

FanoutEmailConsumer.java

  1. @Component
  2. @RabbitListener(bindings = @QueueBinding(
  3. 	value = @Queue(value = "email.topic.queue",durable = "true",antoDelete = "false"),
  4.     exchange = @Exchange(value = "topic_order_exchange",type = "ExchangeTypes.TOPIC")
  5.     key = "#.email.#"
  6. ))
  7. public class TopicEmailConsumer{
  8.     @RabbitHandler
  9.     public void reviceMessage(String message){
  10.         sout("email接收到了的订单信息是："+message);
  11.     }
  12. }

五、RabbitMQ高级

1. 过期时间TTL

概述

过期时间 TTl表示可以对消息设置预期的时间，在这个时间内都可以被消费者接收获取；过了之后消息将自动被删除。RabbitMQ可以对消息和队列设置 TTL，目前有两种方法可以设置

  1. 第一种方法是通过队列属性设置，队列中所有消息都有相同的过期时间

  2. 第二种方法是对消息进行单独设置，每条消息 TTL可以不同

如果上述两种方法同时使用，则消息的过期时间以两者 TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的 TTL值，就称为 dead message被投递到死信队列，消费者将无法再收到该消息

设置队列TTL

 RabbitMqConfiguration.java

  1. @Configuration
  2. public class TTLRabbitMQConfiguration{
  3.     //1.声明注册direct模式的交换机
  4.     @Bean
  5.     public DirectExchange ttldirectExchange(){
  6.         return new DirectExchange("ttl_direct_exchange",true,false);}
  7.     //2.队列的过期时间
  8.     @Bean
  9.     public Queue directttlQueue(){
  10.         //设置过期时间
  11.         Map<String,Object> args = new HashMap<>();
  12.         args.put("x-message-ttl",5000);//这里一定是int类型
  13.         return new Queue("ttl.direct.queue",true,false,false,args);}
  14.     
  15.     @Bean
  16.     public Binding ttlBingding(){
  17.         return BindingBuilder.bin(directttlQueue()).to(ttldirectExchange()).with("ttl");
  18.     }
  19. }

设置消息TTL

  1. public class OrderService{
  2.     @Autowired
  3.     private RabbitTemplate rabbitTemplate;
  4.     //模拟用户下单
  5.     public void makeOrder(String userid,String productid,int num){
  6.         //1.根据商品id查询库存是否足够
  7.         //2.保存订单
  8.         String orderId = UUID.randomUUID().toString();
  9.         sout("订单生产成功："+orderId);
  10.         //3.通过MQ来完成消息的分发
  11.         //参数1：交换机 参数2：路由key/queue队列名称 参数3：消息内容
  12.         String exchangeName = "ttl_order_exchange";
  13.         String routingKey = "ttlmessage";
  14.         //给消息设置过期时间
  15.         MessagePostProcessor messagePostProcessor = new MessagePostProcessor(){
  16.             public Message postProcessMessage(Message message){
  17.                 //这里就是字符串
  18.                 message.getMessageProperties().setExpiration("5000");
  19.                 message.getMessageProperties().setContentEncoding("UTF-8");
  20.                 return message;
  21.             }
  22.         }
  23.         rabbitTemplate.convertAndSend(exchangeName,routingKey,orderId,messagePostProcessor);
  24.     }
  25. }

RabbitMqConfiguration.java

  1. @Configuration
  2. public class TTLRabbitMQConfiguration{
  3.     //1.声明注册direct模式的交换机
  4.     @Bean
  5.     public DirectExchange ttldirectExchange(){
  6.         return new DirectExchange("ttl_direct_exchange",true,false);}
  7.     //2.队列的过期时间
  8.     @Bean
  9.     public Queue directttlQueue(){
  10.         //设置过期时间
  11.         Map<String,Object> args = new HashMap<>();
  12.         args.put("x-message-ttl",5000);//这里一定是int类型
  13.         return new Queue("ttl.direct.queue",true,false,false,args);}
  14.     @Bean
  15.     public Queue directttlMessageQueue(){
  16.         return new Queue("ttlMessage.direct.queue",true,false,false,args);}
  17.     
  18.     @Bean
  19.     public Binding ttlBingding(){
  20.         return BindingBuilder.bin(directttlMessageQueue()).to(ttldirectExchange()).with("ttlmessage");
  21.     }
  22. }

2. 死信队列

概述

DLX，全称 Dead-Letter-Exchange，可以称之为死信交换机，也有人称之为死信邮箱。当消息再一个队列中变成死信之后，它能被重新发送到另一个交换机中，这个交换机就是 DLX，绑定 DLX的队列就称之为死信队列。消息变成死信，可能是由于以下原因：

  1. 消息被拒绝

  2. 消息过期

  3. 队列达到最大长度

DLX也是一个正常的交换机，和一般的交换机没有区别，它能在任何的队列上被指定，实际上就是设置某一个队列的属性，当这个队列中存在死信时，Rabbitmq就会自动地将这个消息重新发布到设置的 DLX上去，进而被路由到另一个队列，即死信队列。

要想使用死信队列，只需要在定义队列的时候设置队列参数x-dead-letter-exchange指定交换机即可

代码

DeadRabbitMqConfiguration.java

  1. @Configuration
  2. public class DeadRabbitMqConfiguration{
  3.     //1.声明注册direct模式的交换机
  4.     @Bean
  5.     public DirectExchange deadDirect(){
  6.         return new DirectExchange("dead_direct_exchange",true,false);}
  7.     //2.队列的过期时间
  8.     @Bean
  9.     public Queue deadQueue(){
  10.         return new Queue("dead.direct.queue",true);}
  11.     @Bean
  12.     public Binding deadbinds(){
  13.         return BindingBuilder.bind(deadDirect()).to(deadQueue()).with("dead");
  14.     }
  15. }

RabbitMqConfiguration.java

  1. @Configuration
  2. public class TTLRabbitMQConfiguration{
  3.     //1.声明注册direct模式的交换机
  4.     @Bean
  5.     public DirectExchange ttldirectExchange(){
  6.         return new DirectExchange("ttl_direct_exchange",true,false);}
  7.     //2.队列的过期时间
  8.     @Bean
  9.     public Queue directttlQueue(){
  10.         //设置过期时间
  11.         Map<String,Object> args = new HashMap<>();
  12.         //args.put("x-max-length",5);
  13.         args.put("x-message-ttl",5000);//这里一定是int类型
  14.         args.put("x-dead-letter-exchange","dead_direct_exchange");
  15.         args.put("x-dead-letter-routing-key","dead");//fanout不需要配置
  16.         return new Queue("ttl.direct.queue",true,false,false,args);}
  17.    
  18.     
  19.     @Bean
  20.     public Binding ttlBingding(){
  21.         return BindingBuilder.bin(directttlQueue()).to(ttldirectExchange()).with("ttlmessage");
  22.     }
  23. }

 

3. 内存磁盘的监控

01 RabbitMQ内存警告

02 RabbitMQ的内存控制

参考帮助文档：http://www.rabbbitmq.com/configure.html

当出现警告的时候，可以通过配置去修改和调整

命令的方式

rabbitmqctl set_vm_memory_high_watermark <fraction>
rabbitmqctl set_vm_memory_high_watermark absolute 50MB

fraction/value 为内存阈值。默认情况是：0.4/2GB，代表的含义是：当 RabbitMQ的内存超过40%时，就会产生警告并且会阻塞所有生产者的连接。通过此命令修改阈值在 Broker重启以后将会失效，通过修改配置文件设置的阈值则不会随着重启而消失，但修改了配置文件一样要重启 Broker才会生效  

配置文件方式 rabbitmq.conf

03 RabbitMQ的内存换页

 

 

04 RabbitMQ的磁盘预警

 

4. 集群（docker集群rabbitmq)

1.先创建三个rabbitmq容器

  1. docker run -itd --name rabbit01 --hostname myrabbit01 -v /home/software/rabbitmqcluster/rabbitmq01:/var/lib/rabbitmq -e RABBITMQ_DEFAULT_USER=chenjinxian -e RABBITMQ_DEFAULT_PASS=chenjinxian  -e RABBITMQ_ERLANG_COOKIE='rabbitmqCookie' -p 15672:15672 -p  5672:5672 -p 25672:25672 rabbitmq:management
  2.  
  3. docker run -itd --name rabbit02 --hostname myrabbit02 -v /home/software/rabbitmqcluster/rabbitmq01:/var/lib/rabbitmq -e RABBITMQ_DEFAULT_USER=chenjinxian -e RABBITMQ_DEFAULT_PASS=chenjinxian  -e RABBITMQ_ERLANG_COOKIE='rabbitmqCookie' --link rabbit01:myrabbit01 -p 15673:15672 -p  5673:5672 -p 25673:25672 rabbitmq:management
  4.  
  5. docker run -itd --name rabbit03 --hostname myrabbit03 -v /home/software/rabbitmqcluster/rabbitmq01:/var/lib/rabbitmq -e RABBITMQ_DEFAULT_USER=chenjinxian -e RABBITMQ_DEFAULT_PASS=chenjinxian  -e RABBITMQ_ERLANG_COOKIE='rabbitmqCookie' --link rabbit01:myrabbit01  --link rabbit02:myrabbit02 -p 15674:15672 -p  5674:5672 -p 25674:25672 rabbitmq:management

启动容器成功后，读者可以访问

 分别通过浏览器访问：ip(自己的IP地址）:15672;ip:15673;ip:15674都可以访问

2.容器节点加入集群

执行如下命令，进入第一个rabbitmq节点容器：

docker exec -it rabbit01 /bin/bash

进入容器后，操作rabbitmq,执行如下命令：

  1. rabbitmqctl stop_app
  2. rabbitmqctl reset
  3. rabbitmqctl start_app
  4. exit

执行如下命令，进入第二个rabbitmq节点容器：

docker exec -it rabbit02 /bin/bash

进入第二个rabbitmq节点容器，执行如下命令：

  1. rabbitmqctl stop_app
  2. rabbitmqctl reset
  3. rabbitmqctl join_cluster --ram rabbit@myrabbit01
  4. rabbitmqctl start_app
  5. exit

，进入第三个rabbitmq节点容器，执行如下命令：

  1. docker exec -it rabbit03 /bin/bash
  2. rabbitmqctl stop_app
  3. rabbitmqctl reset
  4. rabbitmqctl join_cluster --ram rabbit@myrabbit01
  5. rabbitmqctl start_app
  6. exit

最后可以看到节点信息

 

网站建设定制开发 软件系统开发定制 定制软件开发 软件开发定制 定制app开发 app开发定制 app开发定制公司 电商商城定制开发 定制小程序开发 定制开发小程序 客户管理系统开发定制 定制网站 定制开发 crm开发定制 开发公司 小程序开发定制 定制软件 收款定制开发 企业网站定制开发 定制化开发 android系统定制开发 定制小程序开发费用 定制设计 专注app软件定制开发 软件开发定制定制 知名网站建设定制 软件定制开发供应商 应用系统定制开发 软件系统定制开发 企业管理系统定制开发 系统定制开发