RabbitMQ超详学习

• 一、MQ概述
• • 1.MQ的优势
  • 2.MQ的劣势
  • 3.AMQP
  • 4.JMS
  • 5.AMQP 与 JMS 区别
• 二、RabbitMQ介绍
• 三、RabbitMQ安装
• • 方式一：在Linux上安装
  • • 1.安装依赖环境
    • 2.安装Erlang
    • • 2.1解决gblic 版本太低问题
    • 3.安装RabbitMQ
    • 4. 开启管理界面
    • 5.修改配置
    • 6.启动rabbitmq
    • 7.测试
    • 8.注意
  • 方式二：在docker上安装
• 四、控制台的使用：
• • 1.创建用户
  • 2.更新和删除用户
  • 3.添加Virtual Hosts
  • 4.删除 Virtual Hosts
• 五、RabbitMQ入门
• • 1.简单模式
  • • 创建工程
    • 生产者
    • 消费者
  • 2.RabbitMQ工作模式
  • • 2.1Work queues工作队列模式
    • 生产者
    • 消费者1
    • 消费者2
  • 3.Publish/Subscribe发布与订阅模式
  • • 生产者
    • 消费者1
    • 消费者2
  • 4.Routing路由模式
  • • 生产者
    • 消费者1
    • 消费者2
  • 5.Topics通配符模式
  • • 生产者
    • 消费者1
    • 消费者2
• 六、Spring 整合RabbitMQ
• • 1.生产者
  • 2.消费者
  • 3.消息监听器
  • 4.广播监听器
  • 5.（*）星号通配符监听器
  • 6.（#）井号通配符监听器
• 七、Spring Boot整合RabbitMQ
• • 1.生产者
  • 2.消费者
  • 3.消息监听
• 八、RabbitMQ的高级特性
• • 1.消息的可靠投递
  • • 确认模式
    • 退回模式
  • 2.Consumer Ack
  • 3.消费端限流
  • 4.TTL
  • 5.死信队列
  • 6.延迟队列

一、MQ概述

• MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。

• 普通的调用：
  

• 加上RabbitMQ的调用：
  

1.MQ的优势

• 应用解耦：
  系统的耦合性越高，容错性就越低，可维护性就越低。MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦合。提升容错性和可维护性。

• 任务异步处理：
  将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理，提高了应用程序的响应时间。

• 削峰填谷
  

  如系统A，在执行时候会往数据库写数据。但是数据库只能支撑每秒1000左右的并发写入，并发量再高就容易宕机。低峰期的时候并发也就100多个，但是在高峰期时候，并发量会突然激增到5000以上，这个时候数据库肯定卡死了。但是使用了MQ之后，限制消费消息的速度为1000，但是这样一来，高峰期产生的数据势必会被积压在MQ中，高峰就被“削”掉了。但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000QPS，直到消费完积压的消息,这就叫做“填谷”

2.MQ的劣势

• 系统可用性降低：
  系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。
• 系统复杂度提高
  MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。

3.AMQP

AMQP，即 Advanced Message Queuing Protocol（高级消息队列协议），是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，遵循此协议，不收客户端和中间件产品和开发语言限制。2006年，AMQP 规范发布。类比HTTP。

4.JMS

JMS 即 Java 消息服务（JavaMessage Service）应用程序接口，是一个 Java 平台中关于面向消息中间件的API

5.AMQP 与 JMS 区别

• JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
• JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
• JMS规定了两种消息模式；而AMQP的消息模式更加丰富

二、RabbitMQ介绍

• 目前业界有很多的 MQ 产品，例如 RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等，也有直接使用 Redis 充当消息队列的案例，而这些消息队列产品，各有侧重，在实际选型时，需要结合自身需求及 MQ 产品特征，综合考虑。在这里我们选择RabbitMQ

• RabbitMQ 是一个消息代理：它接受和转发消息。您可以将其视为邮局：当您将要邮寄的邮件放入邮箱时，您可以确定邮递员最终会将邮件递送给您的收件人。在这个类比中，RabbitMQ 是一个邮箱、一个邮局和一个邮递员。 RabbitMQ 和邮局之间的主要区别在于它不处理纸张，而是接受、存储和转发二进制数据块 -消息。

• RabbitMQ官方地址

• RabbitMQ 基础架构：
  

名词	概念
Broker	接收和分发消息的应用，RabbitMQ Server就是 Message Broker
Virtual host	出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等
Connection	publisher／consumer 和 broker 之间的 TCP 连接
Channel	如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCPConnection的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销
Exchange	message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) andfanout (multicast)
Queue	消息最终被送到这里等待 consumer 取走
Binding	exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

三、RabbitMQ安装

方式一：在Linux上安装

1.安装依赖环境

#1.在线安装依赖环境：yum install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c++ kernel-devel m4 ncurses-devel tk tc xz

如图：

2.安装Erlang

将这三个文件上传到虚拟机上（没有的小伙伴可以去我的主页里下载）

如图：

# 2.进行安装rpm -ivh erlang-18.3-1.el7.centos.x86_64.rpm

如图：

2.1解决gblic 版本太低问题

当前我们需要使用centos7镜像并且gblic 版本需要2.15，如果出现报错信息：

warning: er1ang-22.0.7-1.e17.x86_64.rpm: Header Vv4 RSA/SHAi signature，key ID 6026dfca: NOKEYerror: Failed dependencies:libc.so.6(GLIBc_2.i4)(64bit) is needed by erlang-22.0.7-1.e17.x86_64libc.so.6(GLIBc_2.15)(64bit) is needed by erlang-22.0.7-1.e17.x86_641ibcrypto.so.10(oPENSSL_1.0.2)(64bit) is'needed by erlang-22.0.7-1.e17.x86_64

说明gblic 版本太低。我们可以查看当前机器的gblic 版本

#1.查看当前机器的gblic 版本strings /lib64/libc.so.6 | grep GLIBC

如果gblic 版本低于2.15就要需要升级glibc版本

#2.使⽤yum更新安装依赖sudo yum install zlib-devel bzip2-devel openssl-devel ncurses-devel sqlitedevel readline-devel tk-devel gcc make -y#3.下载rpm包wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepoel6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-utils-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepoel6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-static-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepoel6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepoel6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-common-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepoel6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-devel-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepoel6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-headers-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepoel6/epel-6-x86_64/glibc-2.17-55.fc20/nscd-2.17-55.el6.x86_64.rpm &#4.安装rpm包sudo rpm -Uvh socat-1.7.3.2-1.1.el7.x86_64.rpm --force --nodeps#5.安装完毕后再查看glibc版本，发现版本已经高于2.15就行了strings /lib64/libc.so.6 | grep GLIBC

3.安装RabbitMQ

输入以下指令：

rpm -ivh socat-1.7.3.2-1.1.el7.x86_64.rpm --force --nodeps

如图：

rpm -ivh rabbitmq-server-3.6.5-1.noarch.rpm

如图：

4. 开启管理界面

输入以下指令：

#打开插件rabbitmq-plugins enable rabbitmq_management

5.修改配置

输入以下指令

vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app

将loopback_users 中的 <>,改为guest
如图：

6.启动rabbitmq

# 启动服务service rabbitmq-server start # 停⽌服务service rabbitmq-server stop # 重启服务service rabbitmq-server restart 

如图：

7.测试

访问 http://ip地址:15672
⽤户名和密码为guest，guest

8.注意

如上图，如果刚装完的小伙伴可能会出现以上提示not found

我们只需执行以下指令：

cd /usr/share/doc/rabbitmq-server-3.6.5/cp rabbitmq.config.example /etc/rabbitmq/rabbitmq.config

如图：

此时便不显示了：

方式二：在docker上安装

#1.打开dockersystemctl start docker#2.搜索镜像docker search rabbitmq

#3.获取镜像docker pull rabbitmq#4.创建容器docker run -d --hostname my-rabbit --name rabbit -p 15672:15672 -p 5673:5672 rabbitmq

#5.查看容器和container iddocker ps

#6.进入容器，注意it后面的是你的刚刚查看的container iddocker exec -it 04610d029f25 /bin/bash#7.进行配置，打开管理界面rabbitmq-plugins enable rabbitmq_management

测试： 端口号为15672
访问：http://IP地址:15672
RabbitMQ默认的账号密码都是guest

到此处说明安装成功了

四、控制台的使用：

1.创建用户

角色	说明
超级管理员(administrator)	可登陆管理控制台，可查看所有的信息，并且可以对⽤户，策略(policy)进⾏操作
监控者(monitoring)	可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使⽤情况，磁盘使⽤情况等)
策略制定者(policymaker)	可登陆管理控制台, 同时可以对policy进⾏管理。但⽆法查看节点的相关信息(上图红框标识的部分)
普通管理者(management)	仅可登陆管理控制台，⽆法看到节点信息，也⽆法对策略进⾏管理
其他	⽆法登陆管理控制台，通常就是普通的⽣产者和消费者

• 点击admin，输入username和password，再点击add user
  
• 如图：可以看到创建的信息
  

2.更新和删除用户

3.添加Virtual Hosts

• 像mysql拥有数据库的概念并且可以指定⽤户对库和表等操作的权限。RabbitMQ也有类似的权限管理；在RabbitMQ中可以虚拟消息服务器Virtual Host，每个Virtual Hosts相当于⼀个相对独⽴的RabbitMQ
  服务器，每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。 相当于mysql的db。Virtual Name⼀般以/开头。

• 如图：进入Virtual Hosts，输入一个以“/”开头的name，点击add Virtual Hosts
  

• 设置Virtual Hosts权限

点击创建好的Virtual Hosts

选择哪些创建好的用户可以访问

如图：Virtual Hosts的访问权限就变了

4.删除 Virtual Hosts

五、RabbitMQ入门

• RabbitMQ提供了6种模式：简单模式，work模式，Publish/Subscribe发布与订阅模式，Routing路由模式，Topics主题模式，RPC远程调用模式

先登录要使用的用户：

1.简单模式

如图：

• 在下图中，“P”是我们的生产者：

  生产无非就是发送，发送消息的程序是生产者。

• “C”是我们的消费者：

  消费与接收具有相似的含义。消费者是一个主要等待接收消息的程序

• 中间的框是一个队列：

  队列是位于 RabbitMQ 中的邮箱的名称。尽管消息流经 RabbitMQ 和您的应用程序，但它们只能存储在队列中。队列仅受主机的内存和磁盘限制，它本质上是一个大的消息缓冲区。许多生产者可以发送去一个队列的消息，许多消费者可以尝试从一个队列接收数据。这就是我们表示队列的方式：，一个 RabbitMQ 代表消费者保存的消息缓冲区

创建工程

• pox.xml

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">    <modelVersion>4.0.0</modelVersion>    <groupId>org.lrk</groupId>    <artifactId>rabbitmq-demo</artifactId>    <version>1.0-SNAPSHOT</version>    <dependencies> <dependency>     <groupId>com.rabbitmq</groupId>     <artifactId>amqp-client</artifactId>     <version>5.6.0</version> </dependency>    </dependencies></project>

• 由于每次都要连接rabbitmq，所以我们抽取连接代码为工具类，每次连接我们只要调工具类就行了：

package com.lrk.rabbitmq.utils;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;import java.util.concurrent.TimeoutException;public class ConnectionUtils {    public static Connection getConnection() throws IOException, TimeoutException { ConnectionFactory connectionFactory = new ConnectionFactory(); //主机地址;默认为 localhost connectionFactory.setHost("192.168.220.129"); //连接端口;默认为 5672 connectionFactory.setPort(5672); //虚拟主机名称;默认为 / connectionFactory.setVirtualHost("/lin"); //连接用户名；默认为guest connectionFactory.setUsername("Link"); //连接密码；默认为guest connectionFactory.setPassword("123"); //创建链接 Connection connection = connectionFactory.newConnection(); return connection;    }}

生产者

package com.lrk.rabbitmq.simple;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Producer {    public static final String QUEUE_NAME = "simple_queue";    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(QUEUE_NAME,true,false,false,null); //发送消息 String messgae="hello"; /**  * 参数1：交换机名称，如果没有指定则使用默认Default Exchage  * 参数2：路由key,简单模式可以传递队列名称  * 参数3：消息其它属性  * 参数4：消息内容  */ channel.basicPublish("",QUEUE_NAME,null,messgae.getBytes()); System.out.println("已发送消息"); //释放资源 channel.close(); connection.close();    }}

执行：

如图：
产生了一个待消费的消息：

点击name进去可以看到发送的消息：

消费者

package com.lrk.rabbitmq.simple;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer {    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null); //接收消息 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {     /**      * 接收到消息执行的回调      * consumerTag 消息者标签，在channel.basicConsume时候可以指定      * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)      * properties 属性信息      * body 消息      */     @Override     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  //路由key  System.out.println("路由key为：" + envelope.getRoutingKey());  //交换机  System.out.println("交换机为：" + envelope.getExchange());  //消息id  System.out.println("消息id为：" + envelope.getDeliveryTag());  //收到的消息  System.out.println("接收到的消息为：" + new String(body, "utf-8"));     } }; /**  * 参数1：队列名称  * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复  会删除消息，设置为false则需要手动确认  * 参数3：消息接收到后回调  */ channel.basicConsume(Producer.QUEUE_NAME,true,defaultConsumer); System.out.println("已发送消息"); //释放资源// channel.close();// connection.close();    }}

执行：

如图：
产生了消费者一个，消息已被消费

2.RabbitMQ工作模式

2.1Work queues工作队列模式

• 工作队列（又名：任务队列）背后的主要思想是避免立即执行资源密集型任务而不得不等待它完成。相反，我们将任务安排在以后完成。我们将任务封装 为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当您运行许多工作人员时，任务将在他们之间共享。
  

• Work Queues 与入门程序的 简单模式 相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。

• 应用场景：对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

生产者

package com.lrk.rabbitmq.work;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Producer {    public static final String QUEUE_NAME = "work_queue";    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(QUEUE_NAME,true,false,false,null); //发送消息 /**  * 参数1：交换机名称，如果没有指定则使用默认Default Exchage  * 参数2：路由key,简单模式可以传递队列名称  * 参数3：消息其它属性  * 参数4：消息内容  */ for (int i = 0; i <30 ; i++) {     String messgae="hello"+i;     channel.basicPublish("",QUEUE_NAME,null,messgae.getBytes());     System.out.println("已发送消息"); } //释放资源 channel.close(); connection.close();    }}

执行：
发送30条消息

如图：

消费者1

package com.lrk.rabbitmq.work;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer1 {    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null); //接收消息 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {     /**      * 接收到消息执行的回调      * consumerTag 消息者标签，在channel.basicConsume时候可以指定      * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)      * properties 属性信息      * body 消息      */     @Override     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  //路由key  System.out.println("路由key为：" + envelope.getRoutingKey());  //交换机  System.out.println("交换机为：" + envelope.getExchange());  //消息id  System.out.println("消息id为：" + envelope.getDeliveryTag());  //收到的消息  System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));     } }; /**  * 参数1：队列名称  * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复  会删除消息，设置为false则需要手动确认  * 参数3：消息接收到后回调  */ channel.basicConsume(Producer.QUEUE_NAME,true,defaultConsumer); //释放资源// channel.close();// connection.close();    }}

消费者2

package com.lrk.rabbitmq.work;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer2 {    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null); //接收消息 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {     /**      * 接收到消息执行的回调      * consumerTag 消息者标签，在channel.basicConsume时候可以指定      * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)      * properties 属性信息      * body 消息      */     @Override     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  //路由key  System.out.println("路由key为：" + envelope.getRoutingKey());  //交换机  System.out.println("交换机为：" + envelope.getExchange());  //消息id  System.out.println("消息id为：" + envelope.getDeliveryTag());  //收到的消息  System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));     } }; /**  * 参数1：队列名称  * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复  会删除消息，设置为false则需要手动确认  * 参数3：消息接收到后回调  */ channel.basicConsume(Producer.QUEUE_NAME,true,defaultConsumer); //释放资源// channel.close();// connection.close();    }}

执行2个消费者：
消费者1，接收了15条，分别是：hello0，hello2，hello4，hello6，…,hello30

消费者2，接收了15条，分别是：hello1，hello3，hello5，hello7，… ,hello29

如图：

3.Publish/Subscribe发布与订阅模式

发布订阅模式： 1、每个消费者监听自己的队列。 2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收 到消息

有几种可用的交换类型：direct、topic、headers 和fanout

生产者

package com.lrk.rabbitmq.ps;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Producer {    public static final String FANOUT_EXCHANGE = "fanout_exchange";    public static final String FANOUT_QUEUE1 = "fanout_queue1";    public static final String FANOUT_QUEUE2 = "fanout_queue2";    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //创建交换机 /**  * 声明交换机  * 参数1：交换机名称  * 参数2：交换机类型，fanout、topic、direct、headers  */ channel.exchangeDeclare(FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(FANOUT_QUEUE1,true,false,false,null); channel.queueDeclare(FANOUT_QUEUE2,true,false,false,null);// 队列绑定交换机 channel.queueBind(FANOUT_QUEUE1,FANOUT_EXCHANGE,""); channel.queueBind(FANOUT_QUEUE2,FANOUT_EXCHANGE,""); //发送消息 /**  * 参数1：交换机名称，如果没有指定则使用默认Default Exchage  * 参数2：路由key,简单模式可以传递队列名称  * 参数3：消息其它属性  * 参数4：消息内容  */ for (int i = 0; i <30 ; i++) {     String messgae="hello"+i;     channel.basicPublish(FANOUT_EXCHANGE,"",null,messgae.getBytes());     System.out.println("已发送消息"); } //释放资源 channel.close(); connection.close();    }}

执行：

消费者1

package com.lrk.rabbitmq.ps;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer1 {    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(Producer.FANOUT_QUEUE1,true,false,false,null); // 队列绑定交换机 channel.queueBind(Producer.FANOUT_QUEUE1,Producer.FANOUT_EXCHANGE,""); //接收消息 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {     /**      * 接收到消息执行的回调      * consumerTag 消息者标签，在channel.basicConsume时候可以指定      * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)      * properties 属性信息      * body 消息      */     @Override     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  //路由key  System.out.println("路由key为：" + envelope.getRoutingKey());  //交换机  System.out.println("交换机为：" + envelope.getExchange());  //消息id  System.out.println("消息id为：" + envelope.getDeliveryTag());  //收到的消息  System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));     } }; /**  * 参数1：队列名称  * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复  会删除消息，设置为false则需要手动确认  * 参数3：消息接收到后回调  */ channel.basicConsume(Producer.FANOUT_QUEUE1,true,defaultConsumer); //释放资源// channel.close();// connection.close();    }}

执行：

消费者2

package com.lrk.rabbitmq.ps;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer2 {    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(Producer.FANOUT_QUEUE2,true,false,false,null); // 队列绑定交换机 channel.queueBind(Producer.FANOUT_QUEUE2,Producer.FANOUT_EXCHANGE,""); //接收消息 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {     /**      * 接收到消息执行的回调      * consumerTag 消息者标签，在channel.basicConsume时候可以指定      * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)      * properties 属性信息      * body 消息      */     @Override     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  //路由key  System.out.println("路由key为：" + envelope.getRoutingKey());  //交换机  System.out.println("交换机为：" + envelope.getExchange());  //消息id  System.out.println("消息id为：" + envelope.getDeliveryTag());  //收到的消息  System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));     } }; /**  * 参数1：队列名称  * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复  会删除消息，设置为false则需要手动确认  * 参数3：消息接收到后回调  */ channel.basicConsume(Producer.FANOUT_QUEUE2,true,defaultConsumer); //释放资源// channel.close();// connection.close();    }}

执行：

绑定：
将交换机和队列绑定在一起

查看创建的交换机：

查看绑定信息：

消费消息：

4.Routing路由模式

• P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
• X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
• C1：消费者，其所在队列指定了需要routing key 为 error 的消息
• C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

生产者

package com.lrk.rabbitmq.routing;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Producer {    public static final String DIRECT_EXCHANGE = "direct_exchange";    public static final String DIRECT_QUEUE1 = "direct_queue1";    public static final String DIRECT_QUEUE2 = "direct_queue2";    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //创建交换机 /**  * 声明交换机  * 参数1：交换机名称  * 参数2：交换机类型，fanout、topic、direct、headers  */ channel.exchangeDeclare(DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(DIRECT_QUEUE1,true,false,false,null); channel.queueDeclare(DIRECT_QUEUE2,true,false,false,null);// 队列绑定交换机 channel.queueBind(DIRECT_QUEUE1,DIRECT_EXCHANGE,"insert"); channel.queueBind(DIRECT_QUEUE2,DIRECT_EXCHANGE,"update"); //发送消息 /**  * 参数1：交换机名称，如果没有指定则使用默认Default Exchage  * 参数2：路由key,简单模式可以传递队列名称  * 参数3：消息其它属性  * 参数4：消息内容  */     String messgae="insert ";     channel.basicPublish(DIRECT_EXCHANGE,"insert",null,messgae.getBytes());     System.out.println("已发送消息"); String messgae2="update "; channel.basicPublish(DIRECT_EXCHANGE,"update",null,messgae2.getBytes()); System.out.println("已发送消息"); //释放资源 channel.close(); connection.close();    }}

执行：

消费者1

package com.lrk.rabbitmq.routing;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer1 {    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(Producer.DIRECT_EXCHANGE,true,false,false,null); // 队列绑定交换机 channel.queueBind(Producer.DIRECT_QUEUE1, Producer.DIRECT_EXCHANGE,"insert"); //接收消息 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {     /**      * 接收到消息执行的回调      * consumerTag 消息者标签，在channel.basicConsume时候可以指定      * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)      * properties 属性信息      * body 消息      */     @Override     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  //路由key  System.out.println("路由key为：" + envelope.getRoutingKey());  //交换机  System.out.println("交换机为：" + envelope.getExchange());  //消息id  System.out.println("消息id为：" + envelope.getDeliveryTag());  //收到的消息  System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));     } }; /**  * 参数1：队列名称  * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复  会删除消息，设置为false则需要手动确认  * 参数3：消息接收到后回调  */ channel.basicConsume(Producer.DIRECT_QUEUE1,true,defaultConsumer); //释放资源// channel.close();// connection.close();    }}

执行：

消费者2

package com.lrk.rabbitmq.routing;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer2 {    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(Producer.DIRECT_QUEUE2,true,false,false,null); // 队列绑定交换机 channel.queueBind(Producer.DIRECT_QUEUE2, Producer.DIRECT_EXCHANGE,"update"); //接收消息 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {     /**      * 接收到消息执行的回调      * consumerTag 消息者标签，在channel.basicConsume时候可以指定      * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)      * properties 属性信息      * body 消息      */     @Override     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  //路由key  System.out.println("路由key为：" + envelope.getRoutingKey());  //交换机  System.out.println("交换机为：" + envelope.getExchange());  //消息id  System.out.println("消息id为：" + envelope.getDeliveryTag());  //收到的消息  System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));     } }; /**  * 参数1：队列名称  * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复  会删除消息，设置为false则需要手动确认  * 参数3：消息接收到后回调  */ channel.basicConsume(Producer.DIRECT_QUEUE2,true,defaultConsumer); //释放资源// channel.close();// connection.close();    }}

执行：

如图：

创建了一个交换机：

查看绑定：

5.Topics通配符模式

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割

通配符规则：

#：匹配一个或多个词*：号匹配不多不少恰好1个词

举例：
item.# ：能够匹配 item.insert.abc 或者 item.insert
item.* ：只能匹配 item.insert

生产者

package com.lrk.rabbitmq.topic;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Producer {    public static final String TOPIC_EXCHANGE = "topic_exchange";    public static final String TOPIC_QUEUE_ALL = "topic_queue_all";    public static final String TOPIC_QUEUE2a_INSERT_UPDATE = "topic_queue_insert_update";    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //创建交换机 /**  * 声明交换机  * 参数1：交换机名称  * 参数2：交换机类型，fanout、topic、direct、headers  */ channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC); //发送消息 /**  * 参数1：交换机名称，如果没有指定则使用默认Default Exchage  * 参数2：路由key,简单模式可以传递队列名称  * 参数3：消息其它属性  * 参数4：消息内容  */ String messgae="insert "; channel.basicPublish(TOPIC_EXCHANGE,"item.insert",null,messgae.getBytes()); System.out.println("已发送消息"); messgae="update "; channel.basicPublish(TOPIC_EXCHANGE,"item.update",null,messgae.getBytes()); System.out.println("已发送消息"); messgae="delete "; channel.basicPublish(TOPIC_EXCHANGE,"item.delete",null,messgae.getBytes()); System.out.println("已发送消息"); //释放资源 channel.close(); connection.close();    }}

执行：

消费者1

package com.lrk.rabbitmq.topic;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer1 {    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(Producer.TOPIC_QUEUE_ALL,true,false,false,null); // 队列绑定交换机 channel.queueBind(Producer.TOPIC_QUEUE_ALL, Producer.TOPIC_EXCHANGE,"item.*"); //接收消息 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {     /**      * 接收到消息执行的回调      * consumerTag 消息者标签，在channel.basicConsume时候可以指定      * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)      * properties 属性信息      * body 消息      */     @Override     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  //路由key  System.out.println("路由key为：" + envelope.getRoutingKey());  //交换机  System.out.println("交换机为：" + envelope.getExchange());  //消息id  System.out.println("消息id为：" + envelope.getDeliveryTag());  //收到的消息  System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));     } }; /**  * 参数1：队列名称  * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复  会删除消息，设置为false则需要手动确认  * 参数3：消息接收到后回调  */ channel.basicConsume(Producer.TOPIC_QUEUE_ALL,true,defaultConsumer); //释放资源// channel.close();// connection.close();    }}

消费者2

package com.lrk.rabbitmq.topic;import com.lrk.rabbitmq.utils.ConnectionUtils;import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer2 {    public static void main(String[] args) throws IOException, TimeoutException { //创建链接 Connection connection = ConnectionUtils.getConnection(); //创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC); //声明（创建）队列 /**  * 参数1：队列名称  * 参数2：是否定义持久化队列  * 参数3：是否独占本次连接  * 参数4：是否在不使用的时候自动删除队列  * 参数5：队列其它参数  */ channel.queueDeclare(Producer.TOPIC_QUEUE2a_INSERT_UPDATE,true,false,false,null); // 队列绑定交换机 channel.queueBind(Producer.TOPIC_QUEUE2a_INSERT_UPDATE, Producer.TOPIC_EXCHANGE,"item.update"); channel.queueBind(Producer.TOPIC_QUEUE2a_INSERT_UPDATE, Producer.TOPIC_EXCHANGE,"item.insert"); //接收消息 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {     /**      * 接收到消息执行的回调      * consumerTag 消息者标签，在channel.basicConsume时候可以指定      * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)      * properties 属性信息      * body 消息      */     @Override     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  //路由key  System.out.println("路由key为：" + envelope.getRoutingKey());  //交换机  System.out.println("交换机为：" + envelope.getExchange());  //消息id  System.out.println("消息id为：" + envelope.getDeliveryTag());  //收到的消息  System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));     } }; /**  * 参数1：队列名称  * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复  会删除消息，设置为false则需要手动确认  * 参数3：消息接收到后回调  */ channel.basicConsume(Producer.TOPIC_QUEUE2a_INSERT_UPDATE,true,defaultConsumer); //释放资源// channel.close();// connection.close();    }}

如图：

查看交换机的绑定：

六、Spring 整合RabbitMQ

1.生产者

• pox.xml

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">    <modelVersion>4.0.0</modelVersion>    <groupId>com.link</groupId>    <artifactId>producer</artifactId>    <version>1.0-SNAPSHOT</version>    <dependencies> <dependency>     <groupId>org.springframework</groupId>     <artifactId>spring-context</artifactId>     <version>5.1.7.RELEASE</version> </dependency> <dependency>     <groupId>org.springframework.amqp</groupId>     <artifactId>spring-rabbit</artifactId>     <version>2.1.8.RELEASE</version> </dependency> <dependency>     <groupId>junit</groupId>     <artifactId>junit</artifactId>     <version>4.12</version> </dependency> <dependency>     <groupId>org.springframework</groupId>     <artifactId>spring-test</artifactId>     <version>5.1.7.RELEASE</version> </dependency>    </dependencies></project>

• 编写配置文件rabbitmq.properties：

配置rabbitmq的IP地址、端口号、账号、密码、虚拟机

rabbitmq.host=192.168.220.129rabbitmq.port=5672rabbitmq.username=Linkrabbitmq.password=123rabbitmq.virtual-host=/lin

• 整合配置文件

<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:context="http://www.springframework.org/schema/context"xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttps://www.springframework.org/schema/context/spring-context.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit.xsd">    <!--加载配置文件-->    <context:property-placeholder location="classpath:properties/rabbitmq.properties"/>    <!-- 定义rabbitmq connectionFactory -->    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"   port="${rabbitmq.port}"   username="${rabbitmq.username}"   password="${rabbitmq.password}"   virtual-host="${rabbitmq.virtual-host}"/>    <!--定义管理交换机、队列-->    <rabbit:admin connection-factory="connectionFactory"/>    <!--定义持久化队列，不存在则自动创建；不绑定到交换机则绑定到默认交换机    默认交换机类型为direct，名字为：""，路由键为队列的名称    -->    <rabbit:queue id="spring_queue" name="spring_queue" auto-declare="true"/>    <!--定义广播交换机中的持久化队列，不存在则自动创建-->    <rabbit:queue id="spring_fanout_queue_1" name="spring_fanout_queue_1" auto-declare="true"/>    <!--定义广播交换机中的持久化队列，不存在则自动创建-->    <rabbit:queue id="spring_fanout_queue_2" name="spring_fanout_queue_2" auto-declare="true"/>    <!--定义广播类型交换机；并绑定上述两个队列-->    <rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange" auto-declare="true"> <rabbit:bindings>     <rabbit:binding queue="spring_fanout_queue_1"/>     <rabbit:binding queue="spring_fanout_queue_2"/> </rabbit:bindings>    </rabbit:fanout-exchange><!--定义广播交换机中的持久化队列，不存在则自动创建-->    <rabbit:queue id="spring_topic_queue_star" name="spring_topic_queue_star" auto-declare="true"/>    <!--定义广播交换机中的持久化队列，不存在则自动创建-->    <rabbit:queue id="spring_topic_queue_well" name="spring_topic_queue_well" auto-declare="true"/>    <!--定义广播交换机中的持久化队列，不存在则自动创建-->    <rabbit:queue id="spring_topic_queue_well2" name="spring_topic_queue_well2" auto-declare="true"/>    <rabbit:topic-exchange id="spring_topic_exchange" name="spring_topic_exchange" auto-declare="true"> <rabbit:bindings>     <rabbit:binding pattern="Link.*" queue="spring_topic_queue_star"/>     <rabbit:binding pattern="Link.#" queue="spring_topic_queue_well"/>     <rabbit:binding pattern="lin.#" queue="spring_topic_queue_well2"/> </rabbit:bindings>    </rabbit:topic-exchange>    <!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/></beans>

• 创建测试文件

package com.link.rabbitmq;import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.test.context.ContextConfiguration;import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;@RunWith(SpringJUnit4ClassRunner.class)@ContextConfiguration(locations = "classpath:/spring/rabbitmq.xml")public class ProducerTest {    @Autowired    private RabbitTemplate rabbitTemplate;    @Test    public void testQueue() { rabbitTemplate.convertAndSend("spring_queue", "只发送spring_queue的消息");    }    /**     * 发送广播     * 交换机类型为 fanout     * 绑定到该交换机的所有队列都能够收到消息     */    @Test    public void fanoutTest(){ /**  * 参数1：交换机名称  * 参数2：路由键名（广播设置为空）  * 参数3：发送的消息内容  */ rabbitTemplate.convertAndSend("spring_fanout_exchange", "", "发送到spring_fanout_exchange交换机的广播消息");    }    /**     * 通配符     * 交换机类型为 topic     * 匹配路由键的通配符，*表示一个单词，#表示多个单词     * 绑定到该交换机的匹配队列能够收到对应消息     */    @Test    public void topicTest(){ /**  * 参数1：交换机名称  * 参数2：路由键名  * 参数3：发送的消息内容  */ rabbitTemplate.convertAndSend("spring_topic_exchange", "hello1", "发送到spring_topic_exchange交换机hello1的消息"); rabbitTemplate.convertAndSend("spring_topic_exchange", "hello2", "发送到spring_topic_exchange交换机hello2的消息"); rabbitTemplate.convertAndSend("spring_topic_exchange", "hello3", "发送到spring_topic_exchange交换机hello3的消息"); rabbitTemplate.convertAndSend("spring_topic_exchange", "hello14", "发送到spring_topic_exchange交换机hello14的消息");    }}

2.消费者

• pox.xml

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">    <modelVersion>4.0.0</modelVersion>    <groupId>com.link</groupId>    <artifactId>consumer</artifactId>    <version>1.0-SNAPSHOT</version>    <dependencies> <dependency>     <groupId>org.springframework</groupId>     <artifactId>spring-context</artifactId>     <version>5.1.7.RELEASE</version> </dependency> <dependency>     <groupId>org.springframework.amqp</groupId>     <artifactId>spring-rabbit</artifactId>     <version>2.1.8.RELEASE</version> </dependency> <dependency>     <groupId>junit</groupId>     <artifactId>junit</artifactId>     <version>4.12</version> </dependency> <dependency>     <groupId>org.springframework</groupId>     <artifactId>spring-test</artifactId>     <version>5.1.7.RELEASE</version> </dependency>    </dependencies></project>

• 编写配置文件rabbitmq.properties：

配置rabbitmq的IP地址、端口号、账号、密码、虚拟机

rabbitmq.host=192.168.220.129rabbitmq.port=5672rabbitmq.username=Linkrabbitmq.password=123rabbitmq.virtual-host=/lin

• 整合配置文件

<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:context="http://www.springframework.org/schema/context"xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttps://www.springframework.org/schema/context/spring-context.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit.xsd">    <!--加载配置文件-->    <context:property-placeholder location="classpath:properties/rabbitmq.properties"/>    <!-- 定义rabbitmq connectionFactory -->    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"   port="${rabbitmq.port}"   username="${rabbitmq.username}"   password="${rabbitmq.password}"   virtual-host="${rabbitmq.virtual-host}"/>    <bean id="springQueueListener" class="com.link.rabbitmq.listener.SpringQueueListener"/>    <bean id="fanoutListener1" class="com.link.rabbitmq.listener.FanoutListener1"/>    <bean id="fanoutListener2" class="com.link.rabbitmq.listener.FanoutListener2"/>    <bean id="topicListenerStar" class="com.link.rabbitmq.listener.TopicListenerStar"/>    <bean id="topicListenerWell" class="com.link.rabbitmq.listener.TopicListenerWell"/>    <bean id="topicListenerWell2" class="com.link.rabbitmq.listener.TopicListenerWell2"/>    <rabbit:listener-container connection-factory="connectionFactory" auto-declare="true"> <rabbit:listener ref="springQueueListener" queue-names="spring_queue"/> <rabbit:listener ref="fanoutListener1" queue-names="spring_fanout_queue_1"/> <rabbit:listener ref="fanoutListener2" queue-names="spring_fanout_queue_2"/> <rabbit:listener ref="topicListenerStar" queue-names="spring_topic_queue_star"/> <rabbit:listener ref="topicListenerWell" queue-names="spring_topic_queue_well"/> <rabbit:listener ref="topicListenerWell2" queue-names="spring_topic_queue_well2"/>    </rabbit:listener-container></beans>

3.消息监听器

public class SpringQueueListener implements MessageListener {    @Override    public void onMessage(Message message) { try {     String msg = new String(message.getBody(), "utf-8");     System.out.printf("接收的路由名称为：%s, 路由键：%s, 队列名： %s, 消息：%s \n",      message.getMessageProperties().getReceivedExchange(),      message.getMessageProperties().getReceivedRoutingKey(),      message.getMessageProperties().getConsumerQueue(),      msg); } catch (Exception e) {     e.printStackTrace(); }    }}

4.广播监听器

public class FanoutListener2 implements MessageListener {    public void onMessage(Message message) { try {     String msg = new String(message.getBody(), "utf-8");     System.out.printf("广播监听器1：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",      message.getMessageProperties().getReceivedExchange(),      message.getMessageProperties().getReceivedRoutingKey(),      message.getMessageProperties().getConsumerQueue(),      msg); } catch (Exception e) {     e.printStackTrace(); }    }}

public class FanoutListener2 implements MessageListener {    public void onMessage(Message message) { try {     String msg = new String(message.getBody(), "utf-8");     System.out.printf("广播监听器2：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",      message.getMessageProperties().getReceivedExchange(),      message.getMessageProperties().getReceivedRoutingKey(),      message.getMessageProperties().getConsumerQueue(),      msg); } catch (Exception e) {     e.printStackTrace(); }    }}

5.（*）星号通配符监听器

public class TopicListenerStar implements MessageListener {    public void onMessage(Message message) { try {     String msg = new String(message.getBody(), "utf-8");     System.out.printf("通配符*监听器：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",      message.getMessageProperties().getReceivedExchange(),      message.getMessageProperties().getReceivedRoutingKey(),      message.getMessageProperties().getConsumerQueue(),      msg); } catch (Exception e) {     e.printStackTrace(); }    }}

6.（#）井号通配符监听器

public class TopicListenerWell implements MessageListener {    public void onMessage(Message message) { try {     String msg = new String(message.getBody(), "utf-8");     System.out.printf("通配符#监听器：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",      message.getMessageProperties().getReceivedExchange(),      message.getMessageProperties().getReceivedRoutingKey(),      message.getMessageProperties().getConsumerQueue(),      msg); } catch (Exception e) {     e.printStackTrace(); }    }}

public class TopicListenerWell2 implements MessageListener {    public void onMessage(Message message) { try {     String msg = new String(message.getBody(), "utf-8");     System.out.printf("通配符#监听器2：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",      message.getMessageProperties().getReceivedExchange(),      message.getMessageProperties().getReceivedRoutingKey(),      message.getMessageProperties().getConsumerQueue(),      msg); } catch (Exception e) {     e.printStackTrace(); }    }}

七、Spring Boot整合RabbitMQ

1.生产者

• pox.xml

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">    <modelVersion>4.0.0</modelVersion>    <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.1.4.RELEASE</version>    </parent>    <groupId>com.link</groupId>    <artifactId>producer</artifactId>    <version>1.0-SNAPSHOT</version>    <dependencies> <dependency>     <groupId>org.springframework.boot</groupId>     <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency>     <groupId>org.springframework.boot</groupId>     <artifactId>spring-boot-starter-test</artifactId> </dependency>    </dependencies></project>

• 启动类

@SpringBootApplicationpublic class ProducerApplication {    public static void main(String[] args) { SpringApplication.run(ProducerApplication.class, args);    }}

• 配置yml文件

spring:  rabbitmq:    host: 192.168.220.129    port: 5672    virtual-host: /lin    username: Link    password: 123

• 绑定交换机和队列

@Configurationpublic class RabbitMQConfig {    //交换机名称    public static final String ITEM_TOPIC_EXCHANGE = "springboot_item_topic_exchange";    //队列名称    public static final String ITEM_QUEUE = "springboot_item_queue";    /**     * 声明交换机     * @return     */    @Bean    public Exchange itemTopicExchange() { return ExchangeBuilder.topicExchange(ITEM_TOPIC_EXCHANGE).durable(true).build();    }    @Bean    public Queue itemQueue() {     return QueueBuilder.durable(ITEM_QUEUE).build();    }    @Bean    public Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue, @Qualifier("itemTopicExchange") Exchange exchange) { return BindingBuilder.bind(queue).to(exchange).with("item.#").noargs();    }}

• 测试

@RunWith(SpringRunner.class)@SpringBootTestpublic class RabbitMQTest {    @Autowired    private RabbitTemplate rabbitTemplate;    @Test    public void test(){ rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.insert", "新增，routing key 为item.insert"); rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.update", "修改，routing key 为item.update"); rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.delete", "删除，routing key 为item.delete");    }}

2.消费者

• pox.xml

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">    <modelVersion>4.0.0</modelVersion>    <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.1.4.RELEASE</version>    </parent>    <groupId>com.link</groupId>    <artifactId>consumer</artifactId>    <version>1.0-SNAPSHOT</version>    <dependencies> <dependency>     <groupId>org.springframework.boot</groupId>     <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency>     <groupId>org.springframework.boot</groupId>     <artifactId>spring-boot-starter-test</artifactId> </dependency>    </dependencies></project>

• 启动类

@SpringBootApplicationpublic class ConsumerApplication {    public static void main(String[] args) { SpringApplication.run(ConsumerApplication.class, args);    }}

• 配置yml文件

spring:  rabbitmq:    host: 192.168.220.129    port: 5672    virtual-host: /lin    username: Link    password: 123

3.消息监听

@Componentpublic class MyListener {    @RabbitListener(queues = "springboot_item_queue")    public void myListener1(String message) { System.out.println("消费者接收到的消息为：" + message);    }}

八、RabbitMQ的高级特性

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我
们提供了两种方式用来控制消息的投递可靠性模式。
confirm 确认模式
return 退回模式
rabbitmq 整个消息投递的路径为：
producer—>rabbitmq broker—>exchange—>queue—>consumer
消息从 producer 到 exchange 则会返回一个 confirmCallback 。
消息从 exchange–>queue 投递失败则会返回一个 returnCallback 。
我们将利用这两个 callback 控制消息的可靠性投递

1.消息的可靠投递

确认模式

 @Test    public void testConfirm() { rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {     /**      *      * @param correlationData 相关配置信息      * @param ack   exchange交换机 是否成功收到了消息。true 成功，false代表失败      * @param cause 失败原因      */     @Override     public void confirm(CorrelationData correlationData, boolean ack, String cause) {  System.out.println("confirm方法被执行了。。。");  if (ack) {      System.out.println("成功接收消息：" + cause);  } else {      System.out.println("接收消息失败：" + cause);      //业务处理  }     } }); rabbitTemplate.convertAndSend("test_exchange_confirm22", "confirm", "message confirm....");    }

退回模式

 @Test    public void testReturn() { //设置交换机处理失败消息的模式 rabbitTemplate.setMandatory(true); //2.设置ReturnCallBack rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {     /**      *      * @param message   消息对象      * @param replyCode 错误码      * @param replyText 错误信息      * @param exchange  交换机      * @param routingKey 路由键      */     @Override     public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {  System.out.println("return 执行了....");  System.out.println(message);  System.out.println(replyCode);  System.out.println(replyText);  System.out.println(exchange);  System.out.println(routingKey);  //处理     } }); //3. 发送消息 rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "message confirm....");    }

2.Consumer Ack

ack指Acknowledge，确认。 表示消费端收到消息后的确认方式。
有三种确认方式：
自动确认：acknowledge=“none”
手动确认：acknowledge=“manual”
根据异常情况确认：acknowledge=“auto”，（这种方式使用麻烦，不作讲解）
其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从
RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么
该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手
动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

@Componentpublic class AckListener implements ChannelAwareMessageListener {    @Override    public void onMessage(Message message, Channel channel) throws Exception { long deliverTag = message.getMessageProperties().getDeliveryTag(); try {     System.out.println(new String(message.getBody()));     System.out.println("处理业务逻辑");     int i = 3/0;//模拟业务处理异常     channel.basicAck(deliverTag, true); } catch (Exception e) {     e.printStackTrace();     //拒绝签收     //requeue：true：表示重回队列     channel.basicNack(deliverTag, true, true);//     channel.basicReject(deliverTag, true); }    }}

3.消费端限流

@Componentpublic class QosListener implements ChannelAwareMessageListener {    @Override    public void onMessage(Message message, Channel channel) throws Exception { Thread.sleep(1000); //1.获取消息 System.out.println(new String(message.getBody())); //2. 处理业务逻辑 //3. 签收 channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);    }}

4.TTL

 <!--ttl-->    <rabbit:queue name="test_queue_ttl" id="test_queue_ttl"> <rabbit:queue-arguments>     <entry key="x-message-ttl" value="100000" value-type="java.lang.Integer"></entry> </rabbit:queue-arguments>    </rabbit:queue>    <rabbit:topic-exchange name="test_exchange_ttl"> <rabbit:bindings>     <rabbit:binding pattern="ttl.#" queue="test_queue_ttl"></rabbit:binding> </rabbit:bindings>    </rabbit:topic-exchange>

@Test    public void testTtl() {// for (int i = 0; i < 10; i++) {//     // 发送消息//     rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....");// }// // 消息后处理对象，设置一些消息的参数信息 MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {     @Override     public Message postProcessMessage(Message message) throws AmqpException {  //1.设置message的信息  message.getMessageProperties().setExpiration("5000");//消息的过期时间  //2.返回该消息  return message;     } };// //消息单独过期// rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....", messagePostProcessor); for (int i = 0; i < 10; i++) {     if(i == 5){  //消息单独过期  rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....",messagePostProcessor);     }else{  //不过期的消息  rabbitTemplate.convertAndSend("test_exchange_ttl", "ttl.hehe", "message ttl....");     } }    }

5.死信队列

    <rabbit:queue name="test_queue_dlx" id="test_queue_dlx"> <!--3. 正常队列绑定死信交换机--> <rabbit:queue-arguments>     <!--3.1 x-dead-letter-exchange：死信交换机名称-->     <entry key="x-dead-letter-exchange" value="exchange_dlx" />     <!--3.2 x-dead-letter-routing-key：发送给死信交换机的routingkey-->     <entry key="x-dead-letter-routing-key" value="dlx.hehe" />     <!--4.1 设置队列的过期时间 ttl-->     <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" />     <!--4.2 设置队列的长度限制 max-length -->     <entry key="x-max-length" value="10" value-type="java.lang.Integer" /> </rabbit:queue-arguments>    </rabbit:queue>    <rabbit:topic-exchange name="test_exchange_dlx"> <rabbit:bindings>     <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx"></rabbit:binding> </rabbit:bindings>    </rabbit:topic-exchange>    <!--2. 声明死信队列(queue_dlx)和死信交换机(exchange_dlx)     -->    <rabbit:queue name="queue_dlx" id="queue_dlx"></rabbit:queue>    <rabbit:topic-exchange name="exchange_dlx"> <rabbit:bindings>     <rabbit:binding pattern="dlx.#" queue="queue_dlx"></rabbit:binding> </rabbit:bindings>    </rabbit:topic-exchange>

生产端测试：

@Test    public void testDlx(){ //1. 测试过期时间，死信消息// rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？"); //2. 测试长度限制后，消息死信// for (int i = 0; i < 20; i++) {//     rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？");// } //3. 测试消息拒收 rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息，我会死吗？");    }

消费端

@Componentpublic class DlxListener implements ChannelAwareMessageListener {    @Override    public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try {     //1.接收转换消息     System.out.println(new String(message.getBody()));     //2. 处理业务逻辑     System.out.println("处理业务逻辑...");     int i = 3/0;//出现错误     //3. 手动签收     channel.basicAck(deliveryTag,true); } catch (Exception e) {     //e.printStackTrace();     System.out.println("出现异常，拒绝接受");     //4.拒绝签收，不重回队列 requeue=false     channel.basicNack(deliveryTag,true,false); }    }}

6.延迟队列

<rabbit:queue id="order_queue" name="order_queue"> <!-- 3. 绑定，设置正常队列过期时间为30分钟--> <rabbit:queue-arguments>     <entry key="x-dead-letter-exchange" value="order_exchange_dlx" />     <entry key="x-dead-letter-routing-key" value="dlx.order.cancel" />     <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" /> </rabbit:queue-arguments>    </rabbit:queue>    <rabbit:topic-exchange name="order_exchange"> <rabbit:bindings>     <rabbit:binding pattern="order.#" queue="order_queue"></rabbit:binding> </rabbit:bindings>    </rabbit:topic-exchange>    <!--  2. 定义死信交换机（order_exchange_dlx）和队列(order_queue_dlx)-->    <rabbit:queue id="order_queue_dlx" name="order_queue_dlx"></rabbit:queue>    <rabbit:topic-exchange name="order_exchange_dlx"> <rabbit:bindings>     <rabbit:binding pattern="dlx.order.#" queue="order_queue_dlx"></rabbit:binding> </rabbit:bindings>    </rabbit:topic-exchange>

生产端测试

@Test    public  void testDelay() throws InterruptedException { //1.发送订单消息。 将来是在订单系统中，下单成功后，发送消息 SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"); rabbitTemplate.convertAndSend("order_exchange","order.msg","订单信息：id=1,time=" + sdf.format(new Date()));    /*//2.打印倒计时10秒    for (int i = 10; i > 0 ; i--) { System.out.println(i+"..."); Thread.sleep(1000);    }*/    }

消费端

@Componentpublic class OrderListener implements ChannelAwareMessageListener {    @Override    public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try {     //1.接收转换消息     System.out.println(new String(message.getBody()));     //2. 处理业务逻辑     System.out.println("处理业务逻辑...");     System.out.println("根据订单id查询其状态...");     System.out.println("判断状态是否为支付成功");     System.out.println("取消订单，回滚库存....");     //3. 手动签收     channel.basicAck(deliveryTag,true); } catch (Exception e) {     //e.printStackTrace();     System.out.println("出现异常，拒绝接受");     //4.拒绝签收，不重回队列 requeue=false     channel.basicNack(deliveryTag,true,false); }    }}

步森服饰商城