学习笔记：RabbitMQ快速入门

---

本文最后更新于 2022-04-30，若内容或图片失效，请留言反馈。部分素材来自网络，若不小心影响到您的利益，请联系博主删除。

---

---

前言

---

• 学习资料：黑马程序员 公众号提供的文档资料链接 | 提取码：dor4）
• 学习视频：黑马程序员 《RabbitMQ全套教程，RabbitMQ消息中间件到实战》
• 写博客的原因：写写笔记，做做记录，巩固知识，方便自己在线阅读。
• 大体上是照搬了公众号提供的资料的。同时对内容进行了一些删减和补充，记录了自己在操作过程中的遇到的问题和解决办法。

---

1.MQ的基本概念

---

1.1.MQ概述

---

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

---

---

• MQ，消息队列，存储消息的中间件
• 分布式系统通信两种方式：直接远程调用和借助第三方完成间接通信
• 发送方称为生产者，接收方称为消费

---

1.2.MQ的优势

---

• 优势：应用解耦、异步提速、削峰填谷

---

1.应用解耦

---

---

系统的耦合性越高，容错性就越低，可维护性就越低。

---

---

使用 MQ 使得应用间解耦，提升容错性和可维护性。

---

2.异步提速

---

---

模拟：一个下单操作耗时：20 + 300 + 300 + 300 = 920ms

用户点击完下单按钮后，需要等待920ms才能得到下单响应，太慢！

---

---

模拟：用户点击完下单按钮后，只需等待25ms就能得到下单响应 (20 + 5 = 25ms)。

提升用户体验和系统吞吐量（单位时间内处理请求的数目）。

---

3.削峰填谷

---

---

---

---

使用了 MQ 之后，限制消费消息的速度为1000，这样一来，高峰期产生的数据势必会被积压在 MQ 中，高峰就被“削”掉了。

但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000，直到消费完积压的消息，这就叫做“填谷”。

使用MQ后，可以提高系统稳定性。

---

4.小结（MQ的优势）

---

• 应用解耦：提高系统容错性和可维护性
• 异步提速：提升用户体验和系统吞吐量
• 削峰填谷：提高系统稳定性

---

1.3.MQ的劣势

---

• 劣势：系统可用性降低、系统复杂度提高、一致性问题

---

---

系统可用性降低

系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用？

系统复杂度提高

MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。

如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？

一致性问题

A 系统处理完业务，通过 MQ 给B、C、D三个系统发消息数据，如果 B 系统、C 系统处理成功，D 系统处理失败。

如何保证消息数据处理的一致性？

---

1.4.使用MQ需要满足的条件

---

既然 MQ 有优势也有劣势，那么使用 MQ 需要满足什么条件呢？

① 生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用，其接口的返回值应该为空，这才让明明下层的动作还没做，上层却当成动作做完了继续往后走，即所谓异步成为了可能。

② 容许短暂的不一致性。

③ 确实是用了有效果。即解耦、提速、削峰这些方面的收益，超过加入MQ，管理MQ这些成本。

---

1.5.常见的MQ产品

---

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

---

---

1.6.RabbitMQ的基本概念

---

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

---

---

2007年，Rabbit 技术公司基于 AMQP 标准开发的 RabbitMQ 1.0 发布。RabbitMQ 采用 Erlang 语言开发。

Erlang 语言由 Ericson 设计，专门为开发高并发和分布式系统的一种语言，在电信领域使用广泛。

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，在消息量大的时候建立 TCP Connection 的开销将是巨大的，效率也较低。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) and fanout (multicast)
• Queue：消息最终被送到这里等待 consumer 取走
• Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

RabbitMQ 提供了 6 种工作模式：简单模式、work queues、Publish/Subscribe 发布与订阅模式、Routing路由模式、Topics 主题模式、RPC 远程调用模式（远程调用，不太算 MQ；暂不作介绍）。

官网对应模式介绍：https://www.rabbitmq.com/getstarted.html

---

---

1.7.JMS

---

• JMS 即 Java 消息服务（JavaMessage Service）应用程序接口，是一个 Java 平台中关于面向消息中间件的 API
• JMS 是 JavaEE 规范中的一种，类比 JDBC
• 很多消息中间件都实现了JMS规范，例如：ActiveMQ。RabbitMQ 官方没有提供 JMS 的实现包，但是开源社区有实现包。

---

1.8.RabbitMQ小结

---

RabbitMQ 是基于 AMQP 协议使用 Erlang 语言开发的一款消息队列产品。

RabbitMQ 提供了6种工作模式，我们学习 5 种。这是今天的重点。

AMQP 是协议，类比 HTTP。

JMS 是 API 规范接口，类比 JDBC。

---

2.RabbitMQ 的安装和配置

---

RabbitMQ 官方网址：http://www.rabbitmq.com

---

2.1.安装RabbitMQ

---

前置操作：关闭防火墙

systemctl disable firewalld.service
1

另外，此处我使用的虚拟机是 CentOS7

---

1.在线安装依赖环境

yum install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c++ kernel-devel m4 ncurses-devel tk tc xz
1

---

2.安装 Erlang

上传资料中给的erlang-18.3-1.el7.centos.x86_64.rpm、socat-1.7.3.2-5.el7.x86_64.rpm、rabbitmq-server-3.6.5-1.noarch.rpm。

其中涉及到的 sftp 操作，可以参考我之前写的博客：《CentOS7的下载安装以及之后的基本操作》的 putty 部分

之后再安装 Erlang

rpm -ivh erlang-18.3-1.el7.centos.x86_64.rpm
1

---

• 如果出现以下错误，则是说明 gblic 的版本太低。

可以使用strings /lib64/libc.so.6 | grep GLIBC这个命令来查看当前版本号

---

---

• 升级 glibc 的操作

使用 yum 更新

sudo yum install zlib-devel bzip2-devel openssl-devel ncurses-devel sqlite-devel readline-devel tk-devel gcc make -y 安装依赖
1

下载rpm包

wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/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/myrepo-el6/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/myrepo-el6/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/myrepo-el6/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/myrepo-el6/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/myrepo-el6/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/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/nscd-2.17-55.el6.x86_64.rpm &
1
2
3
4
5
6
7

sudo rpm -Uvh *-2.17-55.el6.x86_64.rpm --force --nodeps # 安装 rpm 包
1

此时我们可以通过命令 glibc 的版本

strings /lib64/libc.so.6 | grep GLIBC
1

---

3.安装 socat

socat 支持多协议，用于协议处理，端口转发，rabbitmq 依赖于 socat，因此在安装 rabbitmq 前要安装 socat。

rpm -ivh socat-1.7.3.2-5.el7.x86_64.rpm
1

---

4.安装 RabbitMQ

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

---

5.启动、停止、重启

service rabbitmq-server start # 启动服务
1

service rabbitmq-server stop # 停止服务
1

service rabbitmq-server restart # 重启服务
1

---

• 如果出现如下错误

Job for rabbitmq-server.service failed because the control process exited with error code.
See "systemctl status rabbitmq-server.service" and "journalctl -xe" for details.
1
2

可以先通过systemctl status rabbitmq-server.service命令查看情况

---

---

显然问题出在这里：Failed to start LSB: Enable AMQP service provided by RabbitMQ broker

使用如下指令： vim /etc/hosts，在下图位置添加自己的 CentOS7 的 IP 地址 hostname，使用hostname就可以看到自己的主机名

---

---

之后再使用vi /etc/rabbitmq/rabbitmq-env.conf，在文件中添加NODENAME=rabbit@localhost即可。

之后再使用service rabbitmq-server start命令来启动 RabbitMQ 就没有问题了。

---

---

2.2.开启管理界面及配置

---

1.开启管理界面

rabbitmq-plugins enable rabbitmq_management # 开启管理界面
1

2.修改默认配置信息

比如修改密码、配置等等，例如：loopback_users 中的 <<"guest">>，只保留 guest

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

---

---

将图中画框一行改为 {loopback_users, [guest]},

之后我们需要重启一下 RabbitMQ ：service rabbitmq-server restart

在自己的本机中输入：虚拟机的IP地址:默认端口号（一般都是15672）。

可以使用vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app 命令查看

---

---

RabbitMQ 在安装好后，可以访问 http://ip 地址:15672 。其自带了 guest/guest 的用户名和密码。

---

---

如下图的情况，则是进入成功

---

---

在上图中我们发现Config file /etc/rabbitmq/rabbitmq.config(not found)，所以此时我们需要设置rabbitmq.config配置文件

---

3.设置配置文件 rabbitmq.config

cd /usr/share/doc/rabbitmq-server-3.6.5/
1

cp rabbitmq.config.example /etc/rabbitmq/rabbitmq.config
1

---

---

操作完成后就不再出现 not found 的字眼了

---

---

2.3.配置虚拟主机及用户

---

RabbitMQ 在安装好后，可以访问 http://ip地址:15672 。其自带了 guest/guest 的用户名和密码。

---

1.设置用户角色

如果需要创建自定义用户；那么也可以登录管理界面后，进行如下操作：

点击Add a user，输入username和password，并在set下选择角色类型（选择后会自动添加到tag中）

---

---

2.角色说明

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

---

3.Virtual Hosts 配置

类似于 MySQL 拥有数据库的概念并且可以指定用户对库和表等操作的权限。

RabbitMQ 也有类似的权限管理，在 RabbitMQ 中可以虚拟消息服务器 Virtual Host，每个 Virtual Hosts 相当于一个相对独立的 RabbitMQ 服务器，每个 VirtualHost 之间是相互离的。exchange、queue、message不能互通。 相当于 MySQL 的 db。Virtual Name 一般以 / 开头。

创建 Virtual Hosts

点击Add a new virtual host即可

---

---

4.设置 Virtual Hosts 权限

点击上图的 Virtual Hosts （如 /itcast）

我们可以在如下图所示的界面 选择可以管理这个虚拟机的用户，同时我们也可以在此界面 删除管理这个虚拟机的用户

---

---

添加成功

---

---

此时我们也可以使用刚才绑定的用户来登录 RabbitMQ 管理界面（rabbitmq_management）

不过在那之前我们需要先 service rabbitmq-server restart 重启一下

---

---

如上，登录成功。

---

3.RabbitMQ 快速入门

---

3.1.RabbitMQ管理界面介绍

---

当安装完 RabbitMQ，访问安装主机 Ip:15672 即可进入管理界面，初始密码为 guest/guest

---

引用的博客原址：

• https://www.cnblogs.com/Runawayprogrammer/p/15980436.html
• https://www.csdn.net/tags/NtzaYgzsOTczODctYmxvZwO0O0OO0O0O.html

---

---

---

• Overview：概览信息
• Conneticons：连接
• Channels：通道
• Exchanges：交换机
• Queues：队列
• Admin：管理

• connections：无论生产者还是消费者，都需要与 RabbitMQ 建立连接后才可以完成消息的生产和消费，在这里可以查看连接情况。
• channels：通道，建立连接后，会形成通道，消息的投递获取依赖通道。
• Exchanges：交换机，用来实现消息的路由。
• Queues：队列，即消息队列，消息存放在队列中，等待消费，消费后被移除队列。

---

3.1.1.Overview

---

---

• Totals

• Ready： 待消费的消息总数。
• Unacked： 待应答的消息总数。
• Total：总数 Ready + Unacked。

• Publish： producter pub 消息的速率。
• Publisher confirm： broker 确认 pub 消息的速率。
• Deliver(manual ack)： customer 手动确认的速率。
• Deliver( auto ack)： customer 自动确认的速率。
• Consumer ack： customer 正在确认的速率。
• Redelivered： 正在传递 ‘redelivered’ 标志集的消息的速率。
• Get (manual ack)： 响应 basic.get 而要求确认的消息的传输速率。
• Get (auto ack)： 响应于 basic.get 而发送不需要确认的消息的速率。
• Return： 将basic.return发送给producter的速率。
• Disk read： queue 从磁盘读取消息的速率。
• Disk write： queue 从磁盘写入消息的速率。

• Connections：client 的 tcp 连接的总数。
• Channels：通道的总数。
• Exchange：交换器的总数。
• Queues：队列的总数。
• Consumers：消费者的总数。

---

---

• Node

启动一个 broker 就会产生一个 node（节点）。

Nodes 项，显示的是 RabbitMQ 的服务节点,目前有一个本地节点，可以有多个服务节点（比如集群的时候）

• Name：broker名称
• File descriptors：broker 打开的文件描述符和限制。
• Socket descriptors：broker 管理的网络套接字数量和限制。当限制被耗尽时，RabbitMQ将停止接受新的网络连接。
• Erlang processes：erlang 启动的进程数。
• Memory：当前 broker 占用的内存。
• Disk space：当前 broker 占用的硬盘。
• Uptime：当前 broker 持续运行的时长。

---

---

• Ports and contexts

1. AMQP 协议端口（通过 TCP 连接访问该端口，写代码时就是通过这个端口来连接的）
2. 集群端口
3. 管理工具的端口

---

---

• Import / export definitions

导入 / 导出配置文件。比如说将来我们做 RabbitMQ 迁移的时候，我们可以先把配置文件导出去，迁移操作完成之后再导入。

---

---

3.1.2.Connetions

---

当前所有客户端活动的连接。包括生成者和消费者。

在这里可以看客户端连接 RabbitMQ 服务的信息。

• Virtual host：所属的虚拟主机。
• Name：名称。
• User name：使用的用户名。
• State：当前的状态，running：运行中；idle：空闲。
• SSL/TLS：是否使用ssl进行连接。
• Protocol：使用的协议。
• Channels：创建的channel的总数。
• From client：每秒发出的数据包。
• To client：每秒收到的数据包。

---

---

3.1.3.Channels

---

当前连接所有创建的通道。

在这里可以看客户端连接 RabbitMQ 通道的信息。通道是建立在连接之上的。

• channel：名称。
• Virtual host：所属的虚拟主机。
• User name：使用的用户名。
• Mode：渠道保证模式。 可以是以下之一，也可以不是：C: confirm；T：transactional(事务)。
• State ：当前的状态。running ：运行中；idle：空闲。
• Unconfirmed：待 confirm 的消息总数。
• Prefetch：设置的 prefetch 的个数。
• Unacker：待 ack 的消息总数。
• publish：producter pub 消息的速率。
• confirm：producter confirm 消息的速率。
• deliver/get：consumer 获取消息的速率。
• ack：consumer ack 消息的速率。

---

---

3.1.4.Exchanges

---

Exchanges 选项有交换机的信息，并且可以通过 Add a new exchange 来添加交换机。

• Virtual host：所属的虚拟主机。
• Name：名称。
• Type：exchange type
• Features：功能。 可以是以下之一，或者不是：D: 持久化；T：Internal，存在改功能表示这个 exchange 不可以被 client 用来推送消息，仅用来进行 exchange 和 exchange 之间的绑定，否则可以推送消息也可以绑定。
• Message rate in：消息进入的速率。
• Message rate out：消息出去的速率。

---

---

页面添加 exchange 交换机

• virtual host ：选择虚拟机
• Name ：交换机名子
• Type ：交换机类型选择，默认 direct 直连模式，fanout 路由模式吗，topic 模式
• Durability ： 是否需要持久化，true 为持久化
• Auto Delete ：当最后一个绑定到 Exchange上 的队列删除后，自动删除该 Exchange
• Internal ：当前Exchange是否用于 RabbitMQ 内部使用，默认为 False
• Arguments ：扩展参数，用于扩展 AMQP 协议，定制化使用

---

---

3.1.5.Queues

---

Queues 选项有 queue 的信息，并且可以通过 Add a new queue 来添加 queue

• Virtual host：所属的虚拟主机。
• Name：名称。
• Features：功能。 可以是以下之一，或者不是：D: 持久化。
• State：当前的状态，running：运行中；idle：空闲。
• Ready：待消费的消息总数。
• Unacked：待应答的消息总数。
• Total：总数 Ready+Unacked。
• incoming：消息进入的速率。
• deliver/get：消息获取的速率。
• ack：消息应答的速率。

创建队列 queue

• type：此 queue 的类型，默认为 classic 主队列，也可以设置为 quorum 从队列\
• name：此 queue 的名称
• durability：queue 中的消息是否要持久化到硬盘
• auto delete：如果此 queue 没有绑定到任何一个 exchange，是否自动删除此 queue
• arguments：设置一些其它参数

---

---

3.1.6.Admin

---

• Name：名称。
• Tags：角色标签，只能选取一个。
• Can access virtual hosts：允许进入的vhost。
• Has password：设置了密码

角色说明

• administrator （超级管理员）
  • 可登录管理控制台（启用 management plugin 的情况下），可查看所有的信息，并且可以对用户，策略（policy）进行操作。
• monitoring（监控者）
  • 可登录管理控制台（启用 management plugin 的情况下），同时可以查看 rabbitmq 节点的相关信息（进程数，内存使用情况，磁盘使用情况等）
• policymaker（策略制定者）
  • 可登录管理控制台（启用 management plugin 的情况下）， 同时可以对policy进行管理。
• management(普通管理者)
  • 仅可登录管理控制台（启用management plugin的情况下），无法看到节点信息，也无法对策略进行管理。
• none（其他）
  • 无法登录管理控制台，通常就是普通的生产者和消费者。

---

---

3.2.RabbitMQ快速入门

---

3.2.1.需求和步骤

---

• 需求：使用简单模式完成消息传递

• 步骤：
  ① 创建工程（生成者、消费者）
  ② 分别添加依赖
  ③ 编写生产者发送消息
  ④ 编写消费者接收消息

---

---

3.2.2.代码部分（生产者）

---

1.使用 Maven 工具创建如下的项目结构

---

---

2.导入依赖：pom.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.itheima</groupId>
    <artifactId>spring-rabbitmq-producers</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>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

3.Producer_HelloWorld.java

• 步骤：1.创建连接工厂、2.设置参数、3. 创建连接 Connection、4.创建 Channel、5.声明队列 Queue、6.发送消息、7.释放资源

package com.itheima.producer;

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_HelloWorld {
    public static void main(String[] args) throws IOException, TimeoutException {
    
        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        
        //2.设置参数
        factory.setHost("192.168.2.212");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        
        //3.创建连接 Connection
        Connection connection = factory.newConnection();
        
        //4.创建 Channel
        Channel channel = connection.createChannel();
       
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
        参数：
            1. queue：队列名称
            2. durable：是否持久化，(持久化的会把数据写到自带的数据库中)当 mq 重启之后，还在
            3. exclusive：
                * 是否独占。只能有一个消费者监听这队列
                * 当 Connection 关闭时，是否删除队列
            4. autoDelete：是否自动删除。当没有 Consumer 时，自动删除掉
            5. arguments：参数。
        */
        //5.创建队列 Queue
        //如果没有一个名字叫 hello_world 的队列，则会创建该队列；如果有，则不会创建
        channel.queueDeclare("hello_world", true, false, false, null);

        /*
        basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
        参数：
            1. exchange：交换机名称。简单模式下交换机会使用默认的 ""（空字符串）
            2. routingKey：路由名称（简单模式下，routingKey 只要与队列名称相同就可以绑定）
            3. props：配置信息
            4. body：发送消息数据
        */
        String body = "hello RabbitMQ~~~";
        
        //6.发送消息
        channel.basicPublish("", "hello_world", null, body.getBytes());

        //7.释放资源
        channel.close();
        connection.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

---

3.2.3.解析（生产者）

---

运行上面的代码成功后，我们发现一个队列

---

---

点击上面的队列，进入到如下的界面

---

---

这里使用的是默认的交换机（Exchange）

---

---

 //释放资源
channel.close();
connection.close();
1
2
3

因为释放了资源，所以此时是查询不到Connections和Channel的。我们可以注释掉上述两行代码，再次运行程序。

---

---

---

如上面的两图，此时可以发现Connections和Channels。

如下图所示，此时Queues中已经有两条信息（因为程序又执行了一遍）

---

---

3.2.4.代码部分（消费者）

---

1.使用 Maven 工具创建如下的项目结构

---

---

2.导入依赖：pom.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.itheima</groupId>
    <artifactId>rabbitmq-consumer</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <!-- RabbitMQ Java 客户端-->
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.6.0</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

3.Consumer_HelloWorld.java

• 步骤：1.创建连接工厂、2.设置参数、3. 创建连接 Connection、4.创建 Channel、5.声明队列 Queue、6.接收消息

package com.ithiema.consumer;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer_HelloWorld {
    public static void main(String[] args) throws IOException, TimeoutException {

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        
        //2. 设置参数
        factory.setHost("192.168.2.212");//ip  默认值 localhost
        factory.setPort(5672); //端口  默认值 5672
        factory.setVirtualHost("/itcast");//虚拟机 默认值/
        factory.setUsername("heima");//用户名 默认 guest
        factory.setPassword("heima");//密码 默认值 guest
        
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        
        //4. 创建Channel
        Channel channel = connection.createChannel();
        
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
        参数：
            1. queue：队列名称
            2. durable:是否持久化，当 MQ 重启之后，还在
            3. exclusive：
                * 是否独占。只能有一个消费者监听这队列
                * 当 Connection 关闭时，是否删除队列
            4. autoDelete:是否自动删除。当没有 Consumer 时，自动删除掉
            5. arguments：参数。
        */
        //5. 创建队列 Queue
        //之前在书写生产者时已经声明过队列，但下行代码放在这里也没有什么问题
        channel.queueDeclare("hello_world", true, false, false, null);

        Consumer consumer = new DefaultConsumer(channel) {
            /*
                回调方法，当收到消息后，会自动执行该方法
	                1. consumerTag：标识
	                2. envelope：获取一些信息，交换机，路由 key...
	                3. properties:配置信息
	                4. body：数据
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumerTag：" + consumerTag);
                System.out.println("Exchange：" + envelope.getExchange());
                System.out.println("RoutingKey：" + envelope.getRoutingKey());
                System.out.println("properties：" + properties);
                System.out.println("body：" + new String(body));
            }
        };
        
        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        参数：
            1. queue：队列名称
            2. autoAck：是否自动确认
            3. callback：回调对象
        */
        //6. 接收消息
        channel.basicConsume("hello_world", true, consumer);
        
        //关闭资源？不需要
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

以下是控制台上打印的消息

consumerTag：amq.ctag-4x8XsSmEQLu-YIwjDXw0-A
Exchange：
RoutingKey：hello_world
properties：#contentHeader<basic>(content-type=null, content-encoding=null, headers=null,
							 delivery-mode=null, priority=null, correlation-id=null, 
							 reply-to=null, expiration=null, message-id=null, 
							 timestamp=null, type=null, user-id=null, app-id=null,
							 cluster-id=null)
body：hello RabbitMQ~~~
1
2
3
4
5
6
7
8
9

---

3.2.5.小结

---

上述的入门案例中其实使用的是如下的简单模式

---

---

• 在上图的模型中，有以下概念：
  • P：生产者，也就是要发送消息的程序
  • C：消费者：消息的接收者，会一直等待消息到来
  • queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息

---

4.RabbitMQ 的工作模式

---

之前我们已经讲过 “Hello World” 工作模式了，故这里不再赘述。

---

---

4.1.Work queues 工作队列模式

---

模式说明

• Work Queues：与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
• 应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

---

---

代码编写

• Work Queues 与入门程序的简单模式的代码几乎是一样的。可以完全复制，并多复制一个消费者进行多个消费者同时对消费消息的测试。

我们可以模拟一下这个场景

比如在生产者中设置发送五次消息

for (int i = 1; i <= 10; i++) {
    String body = i + "hello rabbitmq~~~";
    //发送消息
    channel.basicPublish("", "work_queues", null, body.getBytes());
}
1
2
3
4
5

同时设置两个消费者接收消息，并且打印真实数据到控制台上

System.out.println("body：" + new String(body));
1

---

---

---

这里是 Queues

---

---

小结

• 在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。

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

例如：短信服务部署多个，只需要有一个节点成功发送即可

---

4.2.Pub/Sub 订阅模式

---

模式说明

在订阅模型中，多了一个 Exchange 角色，而且过程略有变化。

---

---

• P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
• C：消费者，消息的接收者，会一直等待消息到来
• Queue：消息队列，接收消息、缓存消息
• Exchange：交换机（X）。一方面，接收生产者发送的消息。另一方面，知道如何处理消息。
  例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于 Exchange 的类型。
• Exchange 有常见以下3种类型：
  • Fanout：广播，将消息交给所有绑定到交换机的队列
  • Direct：定向，把消息交给符合指定 routing key 的队列
  • Topic：通配符，把消息交给符合 routing pattern（路由模式） 的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力。

因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！

---

代码编写（生产者）

• 步骤：
  • 1.创建连接工厂、2.设置参数、3. 创建连接 Connection、4.创建 Channel（省略）
  • 5.创建交换机、6.创建队列、7.绑定队列和交换机、8.发送消息、9.释放资源

生产者：Producer_PubSub.java

/*
exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable,
		boolean autoDelete, boolean internal, Map<String, Object> arguments)
参数：
    1. exchange:交换机名称
    2. type:交换机类型
        DIRECT("direct")：定向
        FANOUT("fanout")：扇形（广播），发送消息到每一个与之绑定队列。
        TOPIC("topic")：通配符的方式
        HEADERS("headers")：参数匹配

    3. durable：是否持久化
    4. autoDelete：自动删除
    5. internal：内部使用。 一般 false
    6. arguments：参数
*/
String exchangeName = "test_fanout";

//5.创建交换机
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT, true, false, false, null);

//6.创建队列
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
channel.queueDeclare(queue1Name, true, false, false, null);
channel.queueDeclare(queue2Name, true, false, false, null);

/*
queueBind(String queue, String exchange, String routingKey)
参数：
    1. queue：队列名称
    2. exchange：交换机名称
    3. routingKey：路由键，绑定规则
        如果交换机的类型为 fanout ，routingKey 设置为""
*/
//7.绑定队列和交换机
channel.queueBind(queue1Name, exchangeName, "");
channel.queueBind(queue2Name, exchangeName, "");

String body = "日志信息：张三调用了 findAll 方法...日志级别：info ...";

//8.发送消息
channel.basicPublish(exchangeName, "", null, body.getBytes());

//9.释放资源
channel.close();
connection.close();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

---

---

---

---

消费者部分的代码比较简单，与前面没有太大区别（两个消费者分别监听自己的队列），这里就省略掉了。

---

小结

• 交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。
• 发布订阅模式与工作队列模式的区别：
  • 工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机
  • 发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息（底层使用默认交换机）
  • 发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑定到默认的交换机

---

4.3.Routing 路由模式

---

模式说明

• 队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）
• 消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey
• Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的 Routingkey 与消息的 Routing key 完全一致，才会接收到消息

---

---

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

---

代码编写

【生产者】Producer_Routing.java

• 步骤
  • 1.创建连接工厂、2.设置参数、3. 创建连接 Connection、4.创建 Channel
  • 5.创建交换机、6.创建队列、7.绑定队列和交换机、8.发送消息、9.释放资源

//5.创建交换机
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true, false, false, null);
1
2

 //6.创建队列
String queue1Name = "test_direct_queue1";
String queue2Name = "test_direct_queue2";

channel.queueDeclare(queue1Name, true, false, false, null);
channel.queueDeclare(queue2Name, true, false, false, null);
1
2
3
4
5
6

 //7.绑定队列和交换机
 
//队列1 绑定 error
channel.queueBind(queue1Name, exchangeName, "error");
//队列2 绑定 info  error  warning
channel.queueBind(queue2Name, exchangeName, "info");
channel.queueBind(queue2Name, exchangeName, "error");
channel.queueBind(queue2Name, exchangeName, "warning");
1
2
3
4
5
6
7
8

 String body = "日志信息：张三调用了delete方法...出错误了。。。日志级别：error...";

//8.发送消息
channel.basicPublish(exchangeName, "warning", null, body.getBytes());
1
2
3
4

【消费者】 部分与之前的代码无异，更改队列名即可，此处省略。

---

小结

Routing 模式要求队列在绑定交换机时要指定 routing key，消息会转发到符合 routing key 的队列。

---

4.4.Topics 通配符模式

---

模式说明

• Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型 Exchange 可以让队列在绑定 Routing key 的时候使用通配符！
• Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
• 通配符规则：# 匹配一个或多个词，* 匹配不多不少恰好1个词。
  • 例如：item.# 能够匹配 item.insert.abc 或者 item.insert
  • 例如：item.* 只能匹配 item.insert

---

---

• 红色 Queue：绑定的是 usa.# ，因此凡是以 usa. 开头的 routing key 都会被匹配到
• 黄色 Queue：绑定的是 #.news ，因此凡是以 .news 结尾的 routing key 都会被匹配

---

代码部分

【生产者】 需要注意的代码。

channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);
1

// routing key  系统的名称.日志的级别。
// 需求： 所有 error 级别的日志存入数据库，所有 order 系统的日志存入数据库
channel.queueBind(queue1Name,exchangeName,"#.error");
channel.queueBind(queue1Name,exchangeName,"order.*");
channel.queueBind(queue2Name,exchangeName,"*.*");
1
2
3
4
5

【消费者】 部分的代码与之前无异，改变一下队列名即可，此处省略。

---

小结

Topic 主题模式可以实现 Pub/Sub 发布与订阅模式和 Routing 路由模式的功能。

只是 Topic 在配置 routing key 的时候可以使用通配符，显得更加灵活。

---

4.5.工作模式总结

---

• 简单模式 HelloWorld
  • 一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）。

• 工作队列模式 Work Queue
  • 一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）。

• 发布订阅模式 Publish/subscribe
  • 需要设置类型为 fanout 的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列。

• 路由模式 Routing
  • 需要设置类型为 direct 的交换机，交换机和队列进行绑定，并且指定 routing key，当发送消息到交换机后，交换机会根据 routing key 将消息发送到对应的队列。

• 通配符模式 Topic
  • 需要设置类型为 topic 的交换机，交换机和队列进行绑定，并且指定通配符方式的 routing key，当发送消息到交换机后，交换机会根据 routing key 将消息发送到对应的队列。

---

5.Spring 整合 RabbitMQ

---

需求：使用 Spring 整合 RabbitMQ

---

5.1.1.生产端

---

步骤

• 创建生产者工程
• 添加依赖
• 配置整合
• 编写代码发送消息

---

代码部分

1.项目结构

---

---

2.导入依赖

pom.xml

生产者和消费者导入的依赖是相同的

<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>

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.8.0</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
    </plugins>
</build>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

3.配置文件：rabbitmq.properties

rabbitmq.host=192.168.2.212
rabbitmq.port=5672
rabbitmq.username=heima
rabbitmq.password=heima
rabbitmq.virtual-host=/itcast
1
2
3
4
5

4.配置文件：spring-rabbitmq-producer.xml

<?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/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

    <!-- 加载配置文件 -->
    <context:property-placeholder location="classpath: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，名字为：""，路由键为队列的名称
    -->
    <!--
        id：bean 的名称
        name：queue 的名称
        auto-declare：自动创建
        auto-delete：自动删除。 最后一个消费者和该队列断开连接后，自动删除队列
        exclusive：是否独占
        durable：是否持久化
    -->

    <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="heima.*" queue="spring_topic_queue_star"/>
            <rabbit:binding pattern="heima.#" queue="spring_topic_queue_well"/>
            <rabbit:binding pattern="itcast.#" queue="spring_topic_queue_well2"/>
        </rabbit:bindings>
    </rabbit:topic-exchange>

    <!--定义 rabbitTemplate 对象操作可以在代码中方便发送消息-->
    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>

</beans>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

测试类：

package org.example;

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-producer.xml")
public class ProducerTest {
    //注入 RabbitTemplate
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testHelloWorld() {
        //发送消息
        rabbitTemplate.convertAndSend("spring_queue", "hello world spring....");
    }

    /**
     * 发送fanout消息
     */
    @Test
    public void testFanout() {
        //发送消息
        rabbitTemplate.convertAndSend("spring_fanout_exchange", "", "spring fanout....");
    }


    /**
     * 发送topic消息
     */
    @Test
    public void testTopics() {
        //发送消息
        rabbitTemplate.convertAndSend("spring_topic_exchange", "heima.hehe.haha", "spring topic....");
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

---

5.1.2.消费端

---

• 创建生产者工程
• 添加依赖
• 配置整合
• 编写消息监听

---

代码部分

1.项目结构

---

---

2.导入依赖

pom.xml

生产者和消费者导入的依赖是相同的

<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>

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.8.0</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
    </plugins>
</build>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

3.配置文件：rabbitmq.properties

rabbitmq.host=192.168.2.212
rabbitmq.port=5672
rabbitmq.username=heima
rabbitmq.password=heima
rabbitmq.virtual-host=/itcast
1
2
3
4
5

4.配置文件：spring-rabbitmq-consumer.xml

<?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/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    <!--加载配置文件-->
    <context:property-placeholder location="classpath: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="org.example.rabbitmq.listener.SpringQueueListener"/>
    <!--
    <bean id="fanoutListener1" class="org.example.rabbitmq.listener.FanoutListener1"/>
    <bean id="fanoutListener2" class="org.example.rabbitmq.listener.FanoutListener2"/>
    <bean id="topicListenerStar" class="org.example.rabbitmq.listener.TopicListenerStar"/>
    <bean id="topicListenerWell" class="org.example.rabbitmq.listener.TopicListenerWell"/>
    <bean id="topicListenerWell2" class="org.example.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>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

5.SpringQueueListener.class

package org.example.rabbitmq.listener;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;

public class SpringQueueListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        //打印消息
        System.out.println(new String(message.getBody()));
    }
}
1
2
3
4
5
6
7
8
9
10
11
12

6.测试类：ConsumerTest.class

package org.example;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-consumer.xml")
public class ConsumerTest {

    @Test
    public void test1() {
        boolean flag = true;
        while (true) { }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

此处仅仅测试了 Hello World 简单模式，其余的自己创建相关类即可，这里省略。

---

6.SpringBoot 整合 RabbitMQ

---

6.1.生产端

---

步骤

1. 创建生产者 SpringBoot 工程
2. 引入 start，依赖坐标

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1
2
3
4

4. 编写 yml 配置，基本信息配置
5. 定义交换机，队列以及绑定关系的配置类
6. 注入 RabbitTemplate ，调用方法，完成消息发送

---

代码部分

1.创建如下的项目结构

---

---

2.导入依赖：pom.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.itheima</groupId>
    <artifactId>producer-springboot</artifactId>
    <version>0.0.1-SNAPSHOT</version>

    <!-- 父工程依赖 -->
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.4.RELEASE</version>
    </parent>

    <name>product-springboot</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <!-- RabbitMQ -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

3.配置文件：application.yml

# 配置 RabbitMQ 的基本信息
spring:
  rabbitmq:
    host: 192.168.2.212
    port: 5672
    username: guest
    password: guest
    virtual-host: /
1
2
3
4
5
6
7
8

4.核心类：com/ithiema/ProducerApplication.java

package com.ithiema;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class ProducerApplication {
    public static void main(String[] args) {
        SpringApplication.run(ProducerApplication.class);
    }
}
1
2
3
4
5
6
7
8
9
10
11

5.com/ithiema/rabbitmq/config/RabbitMQConfig.java

package com.ithiema.rabbitmq.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    public static final String EXCHANGE_NAME = "boot_topic_exchange";
    public static final String QUEUE_NAME = "boot_queue";

    //1.交换机
    @Bean("bootExchange")
    public Exchange bootExchange() {
        /*
            topicExchange (string name)
            directExchange (string name)
            fanoutExchange (string name)
            headersExchange (string name)
        */
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    }

    //2.Queue 队列
    @Bean("bootQueue")
    public Queue bootQueue() {
        return QueueBuilder.durable(QUEUE_NAME).build();
    }

    //3.队列和交互机绑定关系 Binding
    /*
        1.知道哪个队列
        2.知道哪个交换机
        3.routing key
    */
    @Bean
    public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue, @Qualifier("bootExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

6.测试类：com/itheima/ProducerTest.java

package com.itheima;


import com.ithiema.ProducerApplication;
import com.ithiema.rabbitmq.config.RabbitMQConfig;
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.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@SpringBootTest(classes = ProducerApplication.class)
@RunWith(SpringRunner.class)
public class ProducerTest {

    //1.注入 RabbitTemplate
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSend() {
        //public void convertAndSend(String exchange, String routingKey, Object object)
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, "boot.haha", "boot mq hello~~~");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

---

结果

---

---

6.2.消费端

---

步骤

1. 创建消费者 SpringBoot 工程
2. 引入 start，依赖坐标

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1
2
3
4

3. 编写 yml 配置，基本信息配置
4. 定义监听类，使用 @RabbitListener 注解完成队列监听。

---

代码部分

1.创建如下的项目结构

---

---

2.导入依赖：pom.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
         https://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <groupId>com.itheima</groupId>
    <artifactId>consumer-springboot</artifactId>
    <version>0.0.1-SNAPSHOT</version>

    <!-- 父工程依赖 -->
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.4.RELEASE</version>
    </parent>

    <name>consumer-springboot</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <!-- RabbitMQ -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    </dependencies>

</project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

3.配置文件：application.yml

# 配置 RabbitMQ 的基本信息
spring:
  rabbitmq:
    host: 192.168.2.212
    port: 5672
    username: guest
    password: guest
    virtual-host: /
1
2
3
4
5
6
7
8

4.主类：com/itheima/ConsumerApplication.java

package com.itheima;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class ConsumerApplication {
    public static void main(String[] args) {
        SpringApplication.run(ConsumerApplication.class);
    }
}
1
2
3
4
5
6
7
8
9
10
11

5.com/itheima/RabbitMQListener.java

package com.itheima;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class RabbitMQListener {
    @RabbitListener(queues = "boot_queue")
    public void ListenerQueue(Message message){
        System.out.println(new String(message.getBody()));
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

---

启动主类后的结果

---

---

6.3.小结

---

• SpringBoot 提供了快速整合 RabbitMQ 的方式
• 基本信息再 yml 中配置，队列交互机以及绑定关系在配置类中使用 Bean 的方式配置
• 生产端直接注入 RabbitTemplate 完成消息发送
• 消费端直接使用 @RabbitListener 完成消息接收

---

7.RabbitMQ 高级特性

---

消息可靠性投递、Consumer ACK、消费端限流、TTL、死信队列、延迟队列、日志与监控、消息可靠性分析与追踪、管理（角色、权限等）

---

7.0.项目准备

---

7.0.1.生产端

---

1.项目结构

---

---

2.依赖 pom.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.itheima</groupId>
    <artifactId>rabbitmq-producer-spring</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <java.version>1.8</java.version>
    </properties>

    <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>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

3.配置文件：rabbitmq.properties

rabbitmq.host=192.168.2.212
rabbitmq.port=5672
rabbitmq.username=guest
rabbitmq.password=guest
rabbitmq.virtual-host=/
1
2
3
4
5

4.配置文件：spring-rabbitmq-producer.xml

<?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/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

    <!-- 加载配置文件 -->
    <context:property-placeholder location="classpath: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"/>

    <!-- 定义 rabbitTemplate 对象操作可以在代码中方便发送消息 -->
    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>

</beans>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

5.测试类：com/itheima/test/ProducerTest.java

package com.itheima.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-producer.xml")
public class ProducerTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

---

7.0.2.消费端

---

1.项目结构

---

---

2.导入依赖 pom.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.itheima</groupId>
    <artifactId>rabbitmq-consumer-spring</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <java.version>1.8</java.version>
    </properties>

    <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>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

3.配置文件：rabbitmq.properties

rabbitmq.host=192.168.2.212
rabbitmq.port=5672
rabbitmq.username=guest
rabbitmq.password=guest
rabbitmq.virtual-host=/
1
2
3
4
5

4.配置文件：spring-rabbitmq-producer.xml

<?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/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

    <!-- 加载配置文件 -->
    <context:property-placeholder location="classpath: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}"/>


    <context:component-scan base-package="com.itheima.listener"/>

    <!-- 定义监听器容器 -->
    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1">
        <!-- <rabbit:listener ref="ackListener" queue-names="test_queue_confirm"/>-->
        <!-- <rabbit:listener ref="qosListener" queue-names="test_queue_confirm"/>-->
        <!-- 定义监听器，监听正常队列 -->
        <!--<rabbit:listener ref="dlxListener" queue-names="test_queue_dlx"/>-->
        <!-- 延迟队列效果实现：  一定要监听的是 死信队列！！！ -->
        <!--<rabbit:listener ref="orderListener" queue-names="order_queue_dlx"/>-->
    </rabbit:listener-container>

</beans>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

5.测试类：com/itheima/test/ConsumerTest.java

package com.itheima.test;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-consumer.xml")
public class ConsumerTest {

    @Test
    public void test() {
        while (true) {

        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

---

7.1.消息的可靠投递

---

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任
何消息丢失或者投递失败场景。

RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式：confirm 确认模式、return 退回模式

RabbitMQ 整个消息投递的路径为：producer—>rabbitmq broker—>exchange—>queue—>consumer

• 消息从 producer 到 exchange 则会返回一个 confirmCallback 。
• 消息从 exchange --> queue 投递失败则会返回一个 returnCallback 。

---

消息可靠投递：confirm

---

spring-rabbitmq-producer.xml

<!-- 开启 confirms -->
<rabbit:connection-factory  publisher-confirms="true"/>
1
2

<!-- 消息可靠性投递（生产端） -->
<rabbit:queue id="test_queue_confirm" name="test_queue_confirm"/>

<rabbit:direct-exchange name="test_exchange_confirm">
    <rabbit:bindings>
        <rabbit:binding queue="test_queue_confirm" key="confirm"/>
    </rabbit:bindings>
</rabbit:direct-exchange>
1
2
3
4
5
6
7
8

com/itheima/test/ProducerTest.java

/**
 * 确认模式：
 * 步骤：
 * 1. 确认模式开启：ConnectionFactory 中开启 publisher-confirms="true"
 * 2. 在 rabbitTemplate 定义 ConfirmCallBack 回调函数
 */
@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_confirm", "confirm", "message confirm....");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

---

消息可靠投递：return

---

spring-rabbitmq-producer.xml

<!-- 开启 returns -->
<rabbit:connection-factory  publisher-returns="true"/>
1
2

com/itheima/test/ProducerTest.java

/**
 * 回退模式： 当消息发送给 Exchange 后，Exchange 路由到 Queue 失败是 才会执行 ReturnCallBack
 * 步骤：
 * 		1. 开启回退模式：publisher-returns="true"
 * 		2. 设置 ReturnCallBack
 * 		3. 设置 Exchange 处理消息的模式：
 * 			1. 如果消息没有路由到 Queue ，则丢弃消息（默认）
 * 			2. 如果消息没有路由到 Queue ，返回给消息发送方 ReturnCallBack
 */

@Test
public void testReturn() {

    //设置交换机处理失败消息的模式
    rabbitTemplate.setMandatory(true);

    //设置 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);

            //处理（省略）
        }
    });

    //发送消息
    rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm_666", "message confirm....");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

控制台输出

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.

return 执行了....

(Body:'message confirm....' MessageProperties 
		[headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, 
		receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
312
NO_ROUTE
test_exchange_confirm
confirm_666
1
2
3
4
5
6
7
8
9
10
11
12
13

---

消息的可靠投递：confirm & return 小结

---

我们将利用上述两个 callback 控制消息的可靠性投递

• 设置 ConnectionFactory 的 publisher-confirms=“true” 开启 确认模式。
• 使用 rabbitTemplate.setConfirmCallback 设置回调函数。当消息发送到 exchange 后回调 confirm 方法。在方法中判断 ack ，如果为 true ，则发送成功，如果为 false ，则发送失败，需要处理。
• 设置 ConnectionFactory的publisher-returns=“true” 开启 退回模式。
• 使用 rabbitTemplate.setReturnCallback 设置退回函数，当消息从 exchange 路由到 queue失败后，如果设置了 rabbitTemplate.setMandatory(true) 参数，则会将消息退回给 producer 。并执行回调函数 returnedMessage 。
• 在 RabbitMQ 中也提供了事务机制，但是性能较差，故不做讲解。
• 使用channel下列方法，完成事务控制：
  • txSelect(), 用于将当前channel设置成transaction模式
  • txCommit()，用于提交事务
  • txRollback(),用于回滚事务

---

• 持久化
  • exchange 要持久化
  • queue 要持久化
  • message 要持久化
• 生产方确认 Confirm
• 消费方确认 Ack
• Broker 高可用

---

7.2.Consumer Ack

---

ack（Acknowledge）：表示消费端收到消息后的确认方式。

有三种确认方式：

• 自动确认：acknowledge=“none”
• 手动确认：acknowledge=“manual”
• 根据异常情况确认：acknowledge=“auto”，（这种方式使用麻烦，不作讲解）

其中自动确认是指，当消息一旦被 Consumer 接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。

但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。

如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收。

如果出现异常，则调用 channel.basicNack() 方法，让其自动重新发送消息。

---

spring-rabbitmq-consumer.xml

<!-- 定义监听器容器 -->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual">
    <rabbit:listener ref="ackListener" queue-names="test_queue_confirm"/>
</rabbit:listener-container>
1
2
3
4

com/itheima/listener/AckListener.java

package com.itheima.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

/**
 * Consumer ACK 机制：
 *  1. 设置手动签收。acknowledge="manual"
 *  2. 让监听器类实现 ChannelAwareMessageListener 接口
 *  3. 如果消息成功处理，则调用 channel 的 basicAck() 签收
 *  4. 如果消息处理失败，则调用 channel 的 basicNack() 拒绝签收，broker 重新发送给 consumer
 */

@Component
public class AckListener implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
    	//表示当前收到的一个 tag 的标签
        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();

            //4.拒绝签收
            /*
            第三个参数：requeue：重回队列。如果设置为 true，则消息重新回到 queue，broker 会重新发送该消息给消费端
             */
            channel.basicNack(deliveryTag, true, true);//多条处理
            //channel.basicReject(deliveryTag,true);//单条处理
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

---

• 在 rabbit : listener-container 标签中设置 acknowledge 属性，设置 ack 方式 none：自动确认，manual：手动确认
• 如果在消费端没有出现异常，则调用 channel.basicAck( deliveryTag , false ) ; 方法确认签收消息
• 如果出现异常，则在 catch 中调用 basicNack 或 basicReject，拒绝消息，让 MQ 重新发送消息。

---

7.3.消费端限流

---

• 在<rabbit:listener-container> 中配置 prefetch 属性设置消费端一次拉取多少消息
• 消费端的确认模式一定为手动确认。acknowledge=“manual”

---

---

spring-rabbitmq-consumer.xml

<!-- 定义监听器容器 -->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1">
    <rabbit:listener ref="qosListener" queue-names="test_queue_confirm"/>
</rabbit:listener-container>
1
2
3
4

com/itheima/listener/QosListener.java

package com.itheima.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

/**
 * Consumer 限流机制
 *  1. 确保 ack 机制为手动确认。
 *  2. listener-container 配置属性
 *      perfetch = 1，表示消费端每次从 MQ 拉去一条消息来消费，直到手动确认消费完毕后，才会继续拉去下一条消息。
 */

@Component
public 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);

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

生产端（模拟消息发送）

@Test
public void testSend() {
    for (int i = 0; i < 10; i++) {
        // 发送消息
        rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "message confirm....");
    }
}
1
2
3
4
5
6
7

---

7.4.TTL

---

• TTL 全称 Time To Live（存活时间/过期时间）。
• 当消息到达存活时间后，还没有被消费，会被自动清除。
• RabbitMQ 可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。

---

---

• 设置队列过期时间使用参数：x-message-ttl，单位：ms(毫秒)，会对整个队列消息统一过期。
• 设置消息过期时间使用参数：expiration。单位：ms(毫秒)，当该消息在队列头部时（消费时），会单独判断这一消息是否过期。
• 如果两者都进行了设置，以时间短的为准。

---

spring-rabbitmq-producer.xml

<!-- TTL -->
<rabbit:queue name="test_queue_ttl" id="test_queue_ttl">
    <!-- 设置 queue 的参数 -->
    <rabbit:queue-arguments>
        <!-- x-message-ttl 指队列的过期时间-->
        <entry key="x-message-ttl" value="100000" value-type="java.lang.Integer"/>
    </rabbit:queue-arguments>
</rabbit:queue>

<rabbit:topic-exchange name="test_exchange_ttl">
    <rabbit:bindings>
        <rabbit:binding pattern="ttl.#" queue="test_queue_ttl"/>
    </rabbit:bindings>
</rabbit:topic-exchange>
1
2
3
4
5
6
7
8
9
10
11
12
13
14

/**
 * TTL:过期时间
 * 1.队列统一过期
 * 2.消息单独过期
 * 
 * 如果设置了消息的过期时间，也设置了队列的过期时间，它以时间短的为准。
 * 队列过期后，会将队列所有消息全部移除。
 * 消息过期后，只有消息在队列顶端，才会判断其是否过期(移除掉)
 */
@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....");
        }
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

---

7.5.死信队列

---

死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机）。

当消息成为 Dead message 后，可以被重新发送到另一个交换机，这个交换机就是 DLX。

---

---

消息成为死信的三种情况：

1. 队列消息长度到达限制；
2. 消费者拒接消费消息，basicNack / basicReject，并且不把消息重新放入原目标队列，requeue=false；
3. 原队列存在消息过期设置，消息到达超时时间未被消费；

---

队列绑定死信交换机：

• 给队列设置参数： x-dead-letter-exchange 和 x-dead-letter-routing-key

---

---

生产端：spring-rabbitmq-producer.xml

<!--
    死信队列：
        1. 声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx)
        2. 声明死信队列(queue_dlx)和死信交换机(exchange_dlx)
        3. 正常队列绑定死信交换机
            设置两个参数：
                * x-dead-letter-exchange：死信交换机名称
                * x-dead-letter-routing-key：发送给死信交换机的 routingkey
-->

<!--
    1. 声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx)
-->

<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:bindings>
</rabbit:topic-exchange>


<!--
   2. 声明死信队列(queue_dlx)和死信交换机(exchange_dlx)
-->

<rabbit:queue name="queue_dlx" id="queue_dlx"/>
<rabbit:topic-exchange name="exchange_dlx">
    <rabbit:bindings>
        <rabbit:binding pattern="dlx.#" queue="queue_dlx"/>
    </rabbit:bindings>
</rabbit:topic-exchange>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

生产端：com/itheima/test/ProducerTest.java

/**
 * 发送测试死信消息：
 * 1. 过期时间
 * 2. 长度限制
 * 3. 消息拒收
 */
@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", "我是一条消息，我会死吗？");

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

消费端：com/itheima/listener/DlxListener.java

package com.itheima.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

@Component
public 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);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

消费端：spring-rabbitmq-consumer.xml

<!-- 定义监听器容器 -->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1">
    <!-- 定义监听器，监听正常队列 -->
    <rabbit:listener ref="dlxListener" queue-names="test_queue_dlx"/>
</rabbit:listener-container>
1
2
3
4
5

---

小结

1. 死信交换机和死信队列和普通的没有区别

2. 当消息成为死信后，如果该队列绑定了死信交换机，则消息会被死信交换机重新路由到死信队列

3. 消息成为死信的三种情况：

   1. 队列消息长度到达限制；
   2. 消费者拒接消费消息，并且不重回队列；
   3. 原队列存在消息过期设置，消息到达超时时间未被消费；

---

7.6.延迟队列

---

延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费。

需求：

1. 下单后，30分钟未支付，取消订单，回滚库存。
2. 新用户注册成功7天后，发送短信问候。

实现方式：

1. 定时器
2. 延迟队列

---

---

很可惜，在 RabbitMQ 中并未提供延迟队列功能。

但是可以使用：TTL + 死信队列 组合实现延迟队列的效果。

---

---

spring-rabbitmq-producer.xml

<!--
    延迟队列：
        1. 定义正常交换机（order_exchange）和队列(order_queue)
        2. 定义死信交换机（order_exchange_dlx）和队列(order_queue_dlx)
        3. 绑定，设置正常队列过期时间为30分钟
-->

<!-- 1. 定义正常交换机（order_exchange）和队列(order_queue) -->
<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:bindings>
</rabbit:topic-exchange>

<!--  2. 定义死信交换机（order_exchange_dlx）和 队列（order_queue_dlx） -->
<rabbit:queue id="order_queue_dlx" name="order_queue_dlx"/>
<rabbit:topic-exchange name="order_exchange_dlx">
    <rabbit:bindings>
        <rabbit:binding pattern="dlx.order.#" queue="order_queue_dlx"/>
    </rabbit:bindings>
</rabbit:topic-exchange>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

生产者：com/itheima/test/ProducerTest.java

@Test
public void testDelay() throws InterruptedException {
    //1.发送订单消息。 将来是在订单系统中，下单成功后，发送消息
    rabbitTemplate.convertAndSend("order_exchange", "order.msg", "订单信息：id=1,time=2019年8月17日16:41:47");
    
    /*//2.打印倒计时10秒
    for (int i = 10; i > 0 ; i--) {
        System.out.println(i+"...");
        Thread.sleep(1000);
    }*/
}
1
2
3
4
5
6
7
8
9
10
11

消费者：com/itheima/listener/OrderListener.java

package com.itheima.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

@Component
public 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);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

spring-rabbitmq-consumer.xml

<!-- 定义监听器容器 -->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="1">
    <!-- 延迟队列效果实现：  一定要监听的是 死信队列！！！ -->
    <rabbit:listener ref="orderListener" queue-names="order_queue_dlx"/>
</rabbit:listener-container>
1
2
3
4
5

---

小结

1. 延迟队列 指消息进入队列后，可以被延迟一定时间，再进行消费。
2. RabbitMQ 没有提供延迟队列功能，但是可以使用 ： TTL + DLX 来实现延迟队列效果。

---

7.7.日志与监控

---

7.7.1.RabbitMQ 日志

---

RabbitMQ 默认日志存放路径： /var/log/rabbitmq/rabbit@xxx.log

日志包含了 RabbitMQ 的版本号、Erlang 的版本号、RabbitMQ 服务节点名称、cookie 的 hash 值、RabbitMQ 配置文件地址、内存限制、磁盘限制、默认账户 guest 的创建以及权限配置等等。

---

7.7.2.web 管控台监控

---

此处信息在 3.1.RabbitMQ管理界面介绍 处已介绍的很详细了，故这里不再赘述。

---

7.7.3.rabbitmqctl 管理和监控

---

查看队列

rabbitmqctl list_queues
1

查看exchanges

rabbitmqctl list_exchanges
1

查看用户

rabbitmqctl list_users
1

查看连接

rabbitmqctl list_connections
1

查看消费者信息

rabbitmqctl list_consumers
1

查看环境变量

rabbitmqctl environment
1

查看未被确认的队列

rabbitmqctl list_queues name messages_unacknowledged
1

查看单个队列的内存使用

rabbitmqctl list_queues name memory
1

查看准备就绪的队列

rabbitmqctl list_queues name messages_ready
1

---

7.8.消息追踪

---

在使用任何消息中间件的过程中，难免会出现某条消息异常丢失的情况。对于 RabbitMQ 而言，可能是因为生产者或消费者与 RabbitMQ 断开了连接，而它们与 RabbitMQ 又采用了不同的确认机制；也有可能是因为交换器与队列之间不同的转发策略；甚至是交换器并没有与任何队列进行绑定，生产者又不感知或者没有采取相应的措施；另外 RabbitMQ 本身的集群策略也可能导致消息的丢失。这个时候就需要有一个较好的机制跟踪记录消息的投递过程，以此协助开发和运维人员进行问题的定位。

在 RabbitMQ 中可以使用 Firehose 和 rabbitmq_tracing 插件功能来实现消息追踪。

---

消息追踪-Firehose

firehose 的机制是将生产者投递给 rabbitmq 的消息，rabbitmq 投递给消费者的消息按照指定的格式发送到默认的 exchange 上。

这个默认的exchange的名称为 amq.rabbitmq.trace，它是一个 topic 类型的 exchange。

发送到这个 exchange 上的消息的 routing key 为 publish.exchangename 和 deliver.queuename。

其中 exchangename 和 queuename 为实际 exchange 和 queue 的名称，

分别对应生产者投递到 exchange 的消息，和消费者从 queue 上获取的消息。

注意：打开 trace 会影响消息写入功能，适当打开后请关闭。

• rabbitmqctl trace_on：开启 Firehose 命令

• rabbitmqctl trace_off：关闭 Firehose 命令

---

消息追踪-rabbitmq_tracing

rabbitmq_tracing 和 Firehose 在实现上如出一辙，只不过 rabbitmq_tracing 的方式比 Firehose 多了一层 GUI 的包装，更容易使用和管理。

• 启用插件：rabbitmq-plugins enable rabbitmq_tracing

---

8.RabbitMQ 应用问题

---

• 消息可靠性保障

  • 消息补偿机制

• 消息幂等性保障

  • 乐观锁解决方案

---

消息可靠性保障–消息补偿

需求：100%确保消息发送成功

---

---

消息幂等性保障

幂等性指一次和多次请求某一个资源，对于资源本身应该具有同样的结果。也就是说，其任意多次执行对资源本身所产生的影响均与一次执行的影响相同。

在MQ中指，消费多条相同的消息，得到与消费该消息一次相同的结果。

---

• 乐观锁机制

---

---

9.RabbitMQ 集群搭建

---

RabbitMQ 高可用集群

---

摘要：实际生产应用中都会采用消息队列的集群方案，如果选择RabbitMQ那么有必要了解下它的集群方案原理

一般来说，如果只是为了学习RabbitMQ或者验证业务工程的正确性那么在本地环境或者测试环境上使用其单实例部署就可以了，但是出于MQ中间件本身的可靠性、并发性、吞吐量和消息堆积能力等问题的考虑，在生产环境上一般都会考虑使用RabbitMQ的集群方案。

---

9.1 集群方案的原理

---

RabbitMQ 这款消息队列中间件产品本身是基于 Erlang 编写，Erlang 语言天生具备分布式特性（通过同步 Erlang 集群各节点的 magic cookie 来实现）。因此，RabbitMQ 天然支持 Clustering。这使得 RabbitMQ 本身不需要像 ActiveMQ、Kafka 那样通过 ZooKeeper 分别来实现 HA 方案和保存集群的元数据。集群是保证可靠性的一种方式，同时可以通过水平扩展以达到增加消息吞吐量能力的目的。

---

---

9.2.单机多实例部署

---

由于某些因素的限制，有时候你不得不在一台机器上去搭建一个 rabbitmq 集群，这个有点类似 zookeeper 的单机版。真实生成环境还是要成多机集群的。有关怎么配置多机集群的可以参考其他的资料，这里主要论述如何在单机中配置多个 rabbitmq 实例。

主要参考官方文档：https://www.rabbitmq.com/clustering.html

首先确保 RabbitMQ 运行没有问题

查看 RabbitMQ 状态

rabbitmqctl status
1

停止 RabbitMQ 服务

service rabbitmq-server stop
1

启动第一个节点

RABBITMQ_NODE_PORT=5673 RABBITMQ_NODENAME=rabbit1 rabbitmq-server start
1

---

注意

若执行上条命令出现了这样的错误，且在 /etc/rabbitmq/rabbitmq-env.conf文件中添加NODENAME=rabbit@localhost也于事无补的话

ERROR: epmd error for host 192: badarg (unknown POSIX error)
1

那么这个错误则是因为 hostname 是数字的原因导致的

此时我们需要重新设置 hostname 为英文

hostnamectl set-hostname *随便输入一个英文名称*
1

另外我们还需要在/etc/hosts中添加了 IP 地址 hostname

---

---

重启后，hostname 更改成功，在执行上述命令就没有问题了。

---

---

启动第二个节点

web 管理插件端口占用，所以还要指定其 web 插件占用的端口号

RABBITMQ_NODE_PORT=5674 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15674}]" RABBITMQ_NODENAME=rabbit2 rabbitmq-server start
1

结束命令

rabbitmqctl -n rabbit1 stop
1

rabbitmqctl -n rabbit2 stop
1

rabbit1 操作作为主节点

rabbitmqctl -n rabbit1 stop_app # 停止
1

rabbitmqctl -n rabbit1 reset # 重置	 
1

rabbitmqctl -n rabbit1 start_app
1

rabbit2 操作为从节点

rabbitmqctl -n rabbit2 stop_app
1

rabbitmqctl -n rabbit2 reset
1

rabbitmqctl -n rabbit2 join_cluster rabbit1@'centos7' #''内是主机名换成自己的
1

rabbitmqctl -n rabbit2 start_app
1

查看集群状态

rabbitmqctl cluster_status -n rabbit1
1

web监控

---

---

9.3.集群管理

---

rabbitmqctl join_cluster {cluster_node} [–ram]
1

将节点加入指定集群中。在这个命令执行前需要停止 RabbitMQ 应用并重置节点。

rabbitmqctl cluster_status
1

显示集群的状态。

rabbitmqctl change_cluster_node_type {disc|ram}
1

修改集群节点的类型。在这个命令执行前需要停止 RabbitMQ 应用。

rabbitmqctl forget_cluster_node [–offline]
1

将节点从集群中删除，允许离线执行。

rabbitmqctl update_cluster_nodes {clusternode}
1

在集群中的节点应用启动前咨询 clusternode 节点的最新信息，并更新相应的集群信息。这个和 join_cluster 不同，它不加入集群。

考虑这样一种情况，节点A和节点B都在集群中，当节点A离线了，节点C又和节点B组成了一个集群，然后节点B又离开了集群，当A醒来的时候，它会尝试联系节点B，但是这样会失败，因为节点B已经不在集群中了。

rabbitmqctl cancel_sync_queue [-p vhost] {queue}
1

取消队列 queue 同步镜像的操作。

rabbitmqctl set_cluster_name {name}
1

设置集群名称。集群名称在客户端连接时会通报给客户端。

Federation 和 Shovel 插件也会有用到集群名称的地方。

集群名称默认是集群中第一个节点的名称，通过这个命令可以重新设置。

---

9.4.RabbitMQ镜像集群配置

---

上面已经完成 RabbitMQ 默认集群模式，但并不保证队列的高可用性，尽管交换机、绑定这些可以复制到集群里的任何一个节点，但是队列内容不会复制。虽然该模式解决一项目组节点压力，但队列节点宕机直接导致该队列无法应用，只能等待重启，所以要想在队列节点宕机或故障也能正常应用，就要复制队列内容到集群里的每个节点，必须要创建镜像队列。

镜像队列是基于普通的集群模式的，然后再添加一些策略，所以你还是得先配置普通集群，然后才能设置镜像队列，我们就以上面的集群接着做。

设置的镜像队列可以通过开启的网页的管理端 Admin->Policies，也可以通过命令

rabbitmqctl set_policy my_ha "^" '{"ha-mode":"all"}'
1

---

---

• Name：策略名称
• Pattern：匹配的规则，如果是匹配所有的队列，是 ^.
• Definition：使用 ha-mode 模式中的 all，也就是同步所有匹配的队列。问号链接帮助文档。

---

9.5.负载均衡-HAProxy

---

HAProxy 提供高可用性、负载均衡以及基于 TCP 和 HTTP 应用的代理，支持虚拟主机，它是免费、快速并且可靠的一种解决方案。

包括 Twitter，Reddit，StackOverflow，GitHub 在内的多家知名互联网公司在使用。

HAProxy 实现了一种事件驱动、单一进程模型，此模型支持非常大的并发连接数。

---

9.5.1. 安装HAProxy

---

下载依赖包

yum install gcc vim wget
1

上传资料中提供的 haproxy 源码包，并且解压

tar -zxvf haproxy-1.6.5.tar.gz -C /usr/local
1

进入目录、进行编译、安装

cd /usr/local/haproxy-1.6.5
1

make TARGET=linux31 PREFIX=/usr/local/haproxy
1

make install PREFIX=/usr/local/haproxy
1

mkdir /etc/haproxy
1

赋权

groupadd -r -g 149 haproxy
1

useradd -g haproxy -r -s /sbin/nologin -u 149 haproxy
1

创建 haproxy 配置文件

vim /etc/haproxy/haproxy.cfg
1

---

9.5.2.配置HAProxy

---

配置文件路径：/etc/haproxy/haproxy.cfg

#logging options
global
	log 127.0.0.1 local0 info
	maxconn 5120
	chroot /usr/local/haproxy
	uid 99
	gid 99
	daemon
	quiet
	nbproc 20
	pidfile /var/run/haproxy.pid

defaults
	log global
	
	mode tcp

	option tcplog
	option dontlognull
	retries 3
	option redispatch
	maxconn 2000
	contimeout 5s
   
     clitimeout 60s

     srvtimeout 15s	
#front-end IP for consumers and producters

listen rabbitmq_cluster
	bind 0.0.0.0:5672
	
	mode tcp
	#balance url_param userid
	#balance url_param session_id check_post 64
	#balance hdr(User-Agent)
	#balance hdr(host)
	#balance hdr(Host) use_domain_only
	#balance rdp-cookie
	#balance leastconn
	#balance source //ip
	
	balance roundrobin
	
        server node1 127.0.0.1:5673 check inter 5000 rise 2 fall 2
        server node2 127.0.0.1:5674 check inter 5000 rise 2 fall 2

listen stats
	bind 192.168.2.13:8100
	mode http
	option httplog
	stats enable
	stats uri /rabbitmq-stats
	stats refresh 5s
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

启动 HAproxy 负载

/usr/local/haproxy/sbin/haproxy -f /etc/haproxy/haproxy.cfg
1

查看 haproxy 进程状态

ps -ef | grep haproxy
1

访问如下地址对 MQ 节点进行监控

http://192.168.2.13:8100/rabbitmq-stats
1

代码中访问 MQ 集群地址，则变为访问 haproxy 地址：5672

---

---

com/itheima/test/HelloWorld.java

package com.itheima.test;

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 HelloWorld {
    public static void main(String[] args) throws IOException, TimeoutException {

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost("192.168.2.13");//ip  HaProxy的ip
        factory.setPort(5672); //端口 HaProxy的监听的端口
        //3. 创建连接 Connection
        Connection connection = factory.newConnection();
        //4. 创建Channel
        Channel channel = connection.createChannel();
        //5. 创建队列Queue
        channel.queueDeclare("hello_world", true, false, false, null);
        String body = "hello rabbitmq~~~";
        //6. 发送消息
        channel.basicPublish("", "hello_world", null, body.getBytes());
        //7.释放资源
        channel.close();
        connection.close();

        System.out.println("send success....");

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

---