RabbitMQ详解，用心看完这一篇就够了【重点】

2024-02-17 16:30:22 分类：后端 热度：102 评论： 0

1.1 消息中间件

消息中间件是基于队列与消息传递技术，在网络环境中为应用系统提供同步或异步、可靠的消息传输的支撑性软件系统——百度百科

1.1.1 应用场景

1.1.1.1 异步处理

场景说明：

用户注册后，需要发注册邮件和注册短信，传统的做法有两种

  • 1.串行的方式；
  • 2.并行的方式 ；

(1)串行方式：将注册信息写入数据库后，发送注册邮件，再发送注册短信，以上三个任务全部完成后才返回给客户端。这有一个问题是，邮件，短信并不是必须的，它只是一个通知，而这种做法让客户端等待没有必要等待的东西。

(2)并行方式：将注册信息写入数据库后，发送邮件的同时，发送短信，以上三个任务完成后，返回给客户端，并行的方式能提高处理的时间。

假设三个业务节点分别使用50ms，串行方式使用时间150ms，并行使用时间100ms。虽然并性已经提高的处理时间，但是，前面说过邮件和短信对我正常的使用网站没有任何影响，客户端没有必要等着其发送完成才显示注册成功，应该是写入数据库后就返回。
(3)消息队列
引入消息队列后，把发送邮件，短信不是必须的业务逻辑异步处理。

由此可以看出，引入消息队列后，用户的响应时间就等于写入数据库的时间+写入消息队列的时间(可以忽略不计)，引入消息队列后处理后，响应时间是串行的3倍，是并行的2倍。

1.1.2 应用解耦

场景：

双11是购物狂节，用户下单后，订单系统需要通知库存系统，传统的做法就是订单系统调用库存系统的接口。

这种做法有一个缺点:

  • 当库存系统出现故障时，订单就会失败。
  • 订单系统和库存系统高耦合。

引入消息队列

订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功。

库存系统：订阅下单的消息，获取下单消息，进行库操作。
就算库存系统出现故障,消息队列也能保证消息的可靠投递,不会导致消息丢失。

1.1.3 流量削峰

流量削峰一般在秒杀活动中应用广泛

场景:

秒杀活动，一般会因为流量过大，导致应用挂掉，为了解决这个问题，一般在应用前端加入消息队列。

作用:
1、可以控制活动人数，超过此一定阀值的订单直接丢弃(我为什么秒杀一次都没有成功过呢^^)
2、可以缓解短时间的高流量压垮应用(应用程序按自己的最大处理能力获取订单)

1、用户的请求，服务器收到之后，首先写入消息队列，加入消息队列长度超过最大值，则直接抛弃用户请求或跳转到错误页面。

2、秒杀业务根据消息队列中的请求信息，再做后续处理。

1.1.4 消息队列优缺点

关于消息队列的优点也就是上面列举的，就是在特殊场景下有其对应的好处，解耦、异步、削峰。

缺点有以下几个：

  • 系统可用性降低
    系统引入的外部依赖越多，越容易挂掉。本来你就是 A 系统调用 BCD 三个系统的接口就好了，人 ABCD 四个系统好好的，没啥问题，你偏加个 MQ 进来，万一 MQ 挂了咋整，MQ 一挂，整套系统崩溃的，你不就完了？如何保证消息队列的高可用，可以点击这里查看。

  • 系统复杂度提高
    硬生生加个 MQ 进来，你怎么[保证消息没有重复消费]？怎么[处理消息丢失的情况]？怎么保证消息传递的顺序性？头大头大，问题一大堆，痛苦不已。

  • 一致性问题
    A 系统处理完了直接返回成功了，人都以为你这个请求就成功了；但是问题是，要是 BCD 三个系统那里，BD 两个系统写库成功了，结果 C 系统写库失败了，咋整？你这数据就不一致了。

所以消息队列实际是一种非常复杂的架构，你引入它有很多好处，但是也得针对它带来的坏处做各种额外的技术方案和架构来规避掉，做好之后，你会发现，妈呀，系统复杂度提升了一个数量级，也许是复杂了 10 倍。但是关键时刻，用，还是得用的。

1.2 常用消息中间件

AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有rabbitmq等。

JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。

AMQP和JMS

MQ是消息通信的模型，并发具体实现。现在实现MQ的有两种主流方式：AMQP、JMS。

两者间的区别和联系：

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

常见MQ产品

ActiveMQ：基于JMS
RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
RocketMQ``：基于JMS，阿里巴巴产品，目前交由Apache基金会
Kafka：分布式消息系统，高吞吐量

其实现在主流的消息中间件就4种：kafka、ActiveMQ、RocketMQ、RabbitMQ

下面我们来看一下，他们之间有什么区别，他们分别应该用于什么场景

1.2.1 ActiveMQ

我们先看ActiveMQ。其实一般早些的项目需要引入消息中间件，都是使用的这个MQ，但是现在用的确实不多了，说白了就是有些过时了。我们去它的官网看一看，你会发现官网已经不活跃了，好久才会更新一次。

它的单机吞吐量是万级，一些小的项目已经够用了，但对于高并发的互联网项目完全不够看。

在高可用上，使用的主从架构的实现。
在消息可靠性上，有较低的概率会丢失数据。
综合以上，其实这个产品基本可以弃用掉了，我们完全可以使用RabbitMQ来代替它。

1.2.2 RabbitMQ

RabbitMQ出现后，国内大部分公司都从ActiveMQ切换到了RabbitMQ，基本代替了activeMQ的位置。它的社区还是很活跃的。

它的单机吞吐量也是万级，对于需要支持特别高的并发的情况，它是无法担当重任的。

在高可用上，它使用的是镜像集群模式，可以保证高可用。
在消息可靠性上，它是可以保证数据不丢失的，这也是它的一大优点。

同时它也支持一些消息中间件的高级功能，如：消息重试、死信队列等。

但是，它的开发语言是erlang，国内很少有人精通erlang，所以导致无法阅读源码。
对于大多数中小型公司，不需要面对技术上挑战的情况，使用它还是比较合适的。而对于一些BAT大型互联网公司，显然它就不合适了。

1.2.3 RocketMQ

接下来我们来讨论一下我比较喜欢的MQ-RocketMQ，它是阿里开源的消息中间件，久经沙场，非常靠谱。

它支持高吞吐量，能达到10万级，能承受互联网项目高并发的挑战。

在高可用上，它使用的是分布式架构，可以搭建大规模集群，性能很高。
在消息可靠性上，通过配置，可以保证数据的绝对不丢失。
同时它支持大量的高级功能，如：延迟消息、事务消息、消息回溯、死信队列等等。

它非常适合应用于java系统架构中，因为它使用java语言开发的，我们可以去阅读源码了解更深的底层原理。

目前来看，它没有什么特别的缺点，可以支持高并发下的技术挑战，可以基于它实现分布式事务，大型互联网公司和中小型公司都可以选择使用它来作为消息中间件使用，如果我来做技术选型，我首选的中间件就是它。

1.2.4 Kafka

kafka的吞吐量被公认为中间件中的翘楚，单机可以支持十几万的并发，相当强悍。

在高可用上同样支持分布式集群部署。

在消息可靠性上，如果保证异步的性能，可能会出现消息丢失的情况，因为它保存消息时是先存到磁盘缓冲区的，如果机器出现故障，缓冲区的数据是可能丢失的。

它的功能非常的单一，就是消息的接收与发送，因此不适合应用于许多场景。

它在行业内主要应用于大数据领域，使用它进行用户行为日志的采集和计算，来实现比如“猜你喜欢”的功能。

所以，如果没有大数据的需求，一般不会选择它。

1.2.5 为什么选择RabbitMQ

1、ActiveMQ，性能不是很好，因此在高并发的场景下，直接被pass掉了。它的Api很完善，在中小型互联网公司可以去使用。
2、kafka，主要强调高性能，如果对业务需要可靠性消息的投递的时候。那么就不能够选择kafka了。但是如果做一些日志收集呢，kafka还是很好的。因为kafka的性能是十分好的。
3、RocketMQ，它的特点非常好。它高性能、满足可靠性、分布式事物、支持水平扩展、上亿级别的消息堆积、主从之间的切换等等。MQ的所有优点它基本都满足。但是它最大的缺点：商业版收费。因此它有许多功能是不对外提供的。

1.2.6 比较分析图

1.3 主流消息中间件介绍—RabbitMQ

RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。

1.3.1 特点

RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。
AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

AMQP协议更多用在企业系统内，对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。

RabbitMQ的可靠性是非常好的，数据能够保证百分之百的不丢失。可以使用镜像队列，它的稳定性非常好。所以说在我们互联网的金融行业。对数据的稳定性和可靠性要求都非常高的情况下，我们都会选择RabbitMQ。当然没有kafka性能好，但是要比AvtiveMQ性能要好很多。也可以自己做一些性能的优化。

RabbitMQ可以构建异地双活架构，包括每一个节点存储方式可以采用磁盘或者内存的方式。

1.3.2 RabbitMQ的集群架构


非常经典的 mirror 镜像模式，保证 100% 数据不丢失。在实际工作中也是用得最多的，并且实现非常的简单，一般互联网大厂都会构建这种镜像集群模式。

mirror 镜像队列，目的是为了保证 rabbitMQ 数据的高可靠性解决方案，主要就是实现数据的同步，一般来讲是2 - 3个节点实现数据同步。对于100%数据可靠性解决方案，一般是采用 3 个节点。

如上图所示，用 KeepAlived 做了 HA-Proxy 的高可用，然后有 3 个节点的 MQ 服务，消息发送到主节点上，主节点通过 mirror 队列把数据同步到其他的MQ节点，这样来实现其高可靠。

这就是RabbitMQ整个镜像模式的集群架构。

RabbitMQ集群架构参考：RabbitMQ 的4种集群架构

1.4 安装RabbitMQ

1.4.1 Linux安装

1.4.1.1 更新基本系统

安装任何软件包之前，建议使用以下命令更新软件包和存储库
yum -y update

1.4.1.2 安装Erlang

由于RabbitMQ是基于Erlang（面向高并发的语言）语言开发，所以在安装RabbitMQ之前，需要先安装Erlang。在本教程中我们将安装最新版本的Erlang到服务器中。 Erlang在默认的YUM存储库中不可用，因此您将需要安装EPEL存储库。 运行以下命令相同。

yum -y install epel-release
yum -y update

安装Erlang

yum -y install erlang socat

您现在可以使用以下命令检查Erlang版本。

erl -version

您将得到如下输出：

[root@liptan-pc ~]# erl -version
Erlang (ASYNC_THREADS,HIPE) (BEAM) emulator version 5.10.4
1.4.1.3 安装RabbitMQ

RabbitMQ为预编译并可以直接安装的企业Linux系统提供RPM软件包。 唯一需要的依赖是将Erlang安装到系统中。 我们已经安装了Erlang，我们可以进一步下载RabbitMQ。 通过运行下载Erlang RPM软件包。

1.4.1.3.1 下载RabbitMQ

下载RabbitMQ

wget https://www.rabbitmq.***/releases/rabbitmq-server/v3.6.10/rabbitmq-server-3.6.10-1.el7.noarch.rpm

如果你没有安装wget，可以运行yum -y install wget 。 您可以随时找到最新版本的RabbitMQ下载页面的链接。

1.4.1.3.2 安装RabbitMQ

通过运行导入GPG密钥：

rpm –import https://www.rabbitmq.***/rabbitmq-release-signing-key.asc

运行RPM安装RPM包：

rpm -Uvh rabbitmq-server-3.6.10-1.el7.noarch.rpm

RabbitMQ现已经安装在系统上。

1.4.1.4 使用RabbitMQ

运行

systemctl start rabbitmq-server

开机自启

systemctl enable rabbitmq-server

检查状态

systemctl status rabbitmq-server
1.4.1.5 访问Web控制台
1.4.1.5.1 启动web控制台

启动RabbitMQ Web管理控制台，方法是运行：

rabbitmq-plugins enable rabbitmq_management

通过运行以下命令，将RabbitMQ文件的所有权提供给RabbitMQ用户：

chown -R rabbitmq:rabbitmq /var/lib/rabbitmq/
1.4.1.5.2 创建用户

现在，您将需要为RabbitMQ Web`管理控制台创建管理用户。 运行以下命令相同。

rabbitmqctl add_user admin StrongPassword
rabbitmqctl set_user_tags admin administrator
rabbitmqctl set_permissions -p / admin “.*” “.*” “.*”

将管理员更改为管理员用户的首选用户名。 确保将StrongPassword更改为非常强大的密码。

要访问RabbitMQ的管理面板，请使用您最喜爱的Web浏览器并打开以下URL。

http://Your_Server_IP:15672/
1.4.1.6 管理界面介绍

第一次访问需要登录，默认的账号密码为：guest/guest

主页

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

端口：


下面还是每一个都做一下介绍吧。

1.4.1.6.1 Overview 概要

该栏目主要展示的是MQ的概要信息 , 如消息的数量，Connection，Channel，Exchange，Queue，Consumer的数量

1.4.1.6.2 Exchange 交换器

该栏目主要展示的是当前虚拟主机下的交换器，也可以在此添加一个新的交换器， 并且配 置对应的交换器的规则属性 。

1.4.1.6.3 Queues 队列

该栏目展示的是消息队列的信息，里面有各个队列的概要信息，也可以在此栏目添加队列Queue。

1.4.1.6.4 Admin系统管理

该栏目展示的是用户管理的信息， 包含用户列表的展示，添加用户，添加虚拟主机等信息。

1.4.1.7 添加用户


上面的Tags选项，其实是指定用户的角色，可选的有以下几个：

  • 超级管理员(administrator)
    可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。

  • 监控者(monitoring)
    可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)

  • 策略制定者(policymaker)
    可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。

  • 普通管理者(management)
    仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。

  • 其他
    无法登陆管理控制台，通常就是普通的生产者和消费者。

1.4.1.8 创建虚拟主机（Virtual Hosts）

为了让各个用户可以互不干扰的工作，RabbitMQ添加了虚拟主机（Virtual Hosts）的概念。其实就是一个独立的访问路径，不同用户使用不同路径，各自有自己的队列、交换机，互相不会影响。

创建好虚拟主机，我们还要给用户添加访问权限：

点击添加好的虚拟主机：

进入虚拟主机设置界面：

1.4.2 Windows安装

1.4.2.1 安装erlang

进入erlang的官方下载页面进行下载：erlang下载地址
在下载过程中一定要对应匹配RabbitMQ的版本

双击安装并配置环境变量


1.4.2.2 下载RabbitMQ

RabbitMQ下载地址：RabbitMQ下载地址


双击安装，安装完成后，开始安装RabbitMQ-Plugins插件
先cd D:\software\RabbitMQ\rabbitmq_server-3.8.8\sbin
然后运行命令：rabbitmq-plugins enable rabbitmq_management

执行rabbitmqctl status，出现以下内容，说明成功

然后双击运行rabbitmq-server.bat

进入登录页面，发现启动成功~

然后我们可以将RabbitMQ做成Windows服务
以管理员身份运行cmd

cd D:\software\RabbitMQ\rabbitmq_server-3.8.8\sbin
执行rabbitmq-service.bat install

可以通过任务管理器去查看RabbitMQ服务

以上就是Windows安装RabbitMQ的全部过程，页面设置跟上面的Linux一样。

1.5 RabbitMQ的工作原理介绍

首先先介绍一个简单的一个消息推送到接收的流程，提供一个简单的图：

黄色的圈圈就是我们的消息推送服务，将消息推送到 中间方框里面也就是 rabbitMq的服务器，然后经过服务器里面的交换机、队列等各种关系将数据处理入列后，最终右边的蓝色圈圈消费者获取对应监听的消息。

下图是RabbitMQ的基本结构：

组成部分说明：
Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue
Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费者
Producer：消息生产者，即生产方客户端，生产方客户端将消息发送
Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。

生产者发送消息流程：
1、生产者和Broker建立TCP连接。
2、生产者和Broker建立通道。
3、生产者通过通道消息发送给Broker，由Exchange将消息进行转发。
4、Exchange将消息转发到指定的Queue（队列）

消费者接收消息流程：
1、消费者和Broker建立TCP连接
2、消费者和Broker建立通道
3、消费者监听指定的Queue（队列）
4、当有消息到达Queue时Broker默认将消息推送给消费者。
5、消费者接收到消息。
6、ack回复

1.5.1 RabbitMQ 交换机类型

1.5.1.1 Direct exchange（直连交换机）

直连型交换机（direct exchange）是根据消息携带的路由键（routing key）将消息投递给对应队列的，步骤如下：

1、将一个队列绑定到某个交换机上，同时赋予该绑定一个路由键（routing key）
2、当一个携带着路由值为R的消息被发送给直连交换机时，交换机会把它路由给绑定值同样为R的队列。

1.5.1.2 Fanout exchange（扇型交换机）

扇型交换机（funout exchange）将消息路由给绑定到它身上的所有队列。不同于直连交换机，路由键在此类型上不启任务作用。如果N个队列绑定到某个扇型交换机上，当有消息发送给此扇型交换机时，交换机会将消息的发送给这所有的N个队列

1.5.1.3 Topic exchange（主题交换机）

主题交换机（topic exchanges）中，队列通过路由键绑定到交换机上，然后，交换机根据消息里的路由值，将消息路由给一个或多个绑定队列。

扇型交换机和主题交换机异同：

  • 对于扇型交换机路由键是没有意义的，只要有消息，它都发送到它绑定的所有队列上
  • 对于主题交换机，路由规则由路由键决定，只有满足路由键的规则，消息才可以路由到对应的队列上

介绍一下规则

* (星号) 用来表示一个单词 (必须出现的)
# (井号) 用来表示任意数量（零个或多个）单词

通配的绑定键是跟队列进行绑定的，举个小例子
队列Q1 绑定键为*.TT.* 队列Q2绑定键为 TT.#
如果一条消息携带的路由键为 A.TT.B，那么队列Q1将会收到；
如果一条消息携带的路由键为TT.AA.BB，那么队列Q2将会收到；

主题交换机的强大之处

当一个队列的绑定键为 #（井号） 的时候，这个队列将会无视消息的路由键，接收所有的消息。

当* (星号) 和#(井号) 这两个特殊字符都未在绑定键中出现的时候，此时主题交换机就拥有的直连交换机的行为。
所以主题交换机也就实现了扇形交换机的功能，和直连交换机的功能。

1.6 RabbitMQ使用

本次实例教程采用springboot版本为 2.1.7.RELEASE，需要创建2个springboot项目，一个 rabbitmq-provider （生产者），一个rabbitmq-consumer（消费者）

首先创建 rabbitmq-provider
pom.xml里导入相关的依赖：

        <!--rabbitmq-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

然后编写application.yml配置文件：

ps：里面的虚拟host配置项不是必须的，我自己在rabbitmq服务上创建了自己的虚拟host，所以我配置了；你们不创建，就不用加这个配置项。

server:
  port: 8021
spring:
  #给项目来个名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 服务器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虚拟host 可以不设置,使用server默认host
    virtual-host: J******Host

1.6.1 直连交换机（direct exchange）

接着我们先使用下direct exchange(直连型交换机)，创建DirectRabbitConfig.java（对于队列和交换机持久化以及连接使用设置，在注释里有说明，后面的不同交换机的配置就不做同样说明了）：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class DirectRabbitConfig {
 
    //队列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        // return new Queue("TestDirectQueue",true,true,false);
 
        //一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue("TestDirectQueue",true);
    }
 
    //Direct交换机 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
      //  return new DirectExchange("TestDirectExchange",true,true);
        return new DirectExchange("TestDirectExchange",true,false);
    }
 
    //绑定  
    //将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
     
    @Bean
    DirectExchange lonelyDirectExchange() {
        return new DirectExchange("lonelyDirectExchange");
    }
}

然后写个简单的接口进行消息推送（根据需求也可以改为定时任务等等，具体看需求），SendMessageController.java：

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@RestController
public class SendMessageController {

    //使用RabbitTemplate,这提供了接收/发送等等方法
    @Autowired
    RabbitTemplate rabbitTemplate;  
 
    @GetMapping("/sendDirectMessage")
    public String sendDirectMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> map=new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return "ok";
    }
}

把rabbitmq-provider项目运行，调用下接口：

因为我们目前还没弄消费者 rabbitmq-consumer，消息没有被消费的，我们去rabbitMq管理页面看看，是否推送成功：

再看看队列：

很好，消息已经推送到rabbitMq服务器上面了。

接下来，创建rabbitmq-consumer项目：

添加pom.xml里的Maven依赖：

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

然后是 application.yml：

server:
  port: 8022
spring:
  #给项目来个名字
  application:
    name: rabbitmq-consumer
  #配置rabbitMq 服务器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虚拟host 可以不设置,使用server默认host
    virtual-host: J******Host

然后一样，创建DirectRabbitConfig.java（消费者单纯的使用，其实可以不用添加这个配置，直接建后面的监听就好，使用注解来让监听器监听对应的队列即可。配置上了的话，其实消费者也是生成者的身份，也能推送该消息）：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DirectRabbitConfig {
 
    //队列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        return new Queue("TestDirectQueue",true);
    }
 
    //Direct交换机 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("TestDirectExchange");
    }
 
    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }
}

然后是创建消息接收监听类，DirectReceiver.java：

@***ponent
//监听的队列名称 TestDirectQueue
@RabbitListener(queues = "TestDirectQueue")
public class DirectReceiver {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("DirectReceiver消费者收到消息  : " + testMessage.toString());
    }
 
}

然后将rabbitmq-consumer项目运行起来，可以看到把之前推送的那条消息消费下来了：

然后可以再继续调用rabbitmq-provider项目的推送消息接口，可以看到消费者即时消费消息：

那么直连交换机既然是一对一，那如果咱们配置多台监听绑定到同一个直连交互的同一个队列，会怎么样？

可以看到是实现了轮询的方式对消息进行消费，而且不存在重复消费。

1.6.2 主题交换机（Topic Exchange）

接着，我们使用Topic Exchange主题交换机。

在rabbitmq-provider项目里面创建TopicRabbitConfig.java：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class TopicRabbitConfig {
    //绑定键
    public final static String man = "topic.man";
    public final static String woman = "topic.woman";
 
    @Bean
    public Queue firstQueue() {
        return new Queue(TopicRabbitConfig.man);
    }
 
    @Bean
    public Queue secondQueue() {
        return new Queue(TopicRabbitConfig.woman);
    }
 
    @Bean
    TopicExchange exchange() {
        return new TopicExchange("topicExchange");
    }
 
    //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
    //这样只要是消息携带的路由键是topic.man,才会分发到该队列
    @Bean
    Binding bindingExchangeMessage() {
        return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
    }
 
    //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
    // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
    @Bean
    Binding bindingExchangeMessage2() {
        return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
    }
 
}

然后添加多2个接口，用于推送消息到主题交换机：

    @GetMapping("/sendTopicMessage1")
    public String sendTopicMessage1() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: M A N ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> manMap = new HashMap<>();
        manMap.put("messageId", messageId);
        manMap.put("messageData", messageData);
        manMap.put("createTime", createTime);
        rabbitTemplate.convertAndSend("topicExchange", "topic.man", manMap);
        return "ok";
    }
 
    @GetMapping("/sendTopicMessage2")
    public String sendTopicMessage2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: woman is all ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> womanMap = new HashMap<>();
        womanMap.put("messageId", messageId);
        womanMap.put("messageData", messageData);
        womanMap.put("createTime", createTime);
        rabbitTemplate.convertAndSend("topicExchange", "topic.woman", womanMap);
        return "ok";
    }
}

生产者这边已经完事，先不急着运行，在rabbitmq-consumer项目上，创建TopicManReceiver.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.***ponent;
import java.util.Map;

@***ponent
@RabbitListener(queues = "topic.man")
public class TopicManReceiver {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicManReceiver消费者收到消息  : " + testMessage.toString());
    }
}

再创建一个TopicTotalReceiver.java：

package ***.elegant.rabbitmqconsumer.receiver;
 
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.***ponent;
import java.util.Map;

@***ponent
@RabbitListener(queues = "topic.woman")
public class TopicTotalReceiver {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicTotalReceiver消费者收到消息  : " + testMessage.toString());
    }
}

同样，加主题交换机的相关配置，TopicRabbitConfig.java（消费者一定要加这个配置吗？ 不需要的其实，理由在前面已经说过了）：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class TopicRabbitConfig {
    //绑定键
    public final static String man = "topic.man";
    public final static String woman = "topic.woman";
 
    @Bean
    public Queue firstQueue() {
        return new Queue(TopicRabbitConfig.man);
    }
 
    @Bean
    public Queue secondQueue() {
        return new Queue(TopicRabbitConfig.woman);
    }
 
    @Bean
    TopicExchange exchange() {
        return new TopicExchange("topicExchange");
    }
 
    //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
    //这样只要是消息携带的路由键是topic.man,才会分发到该队列
    @Bean
    Binding bindingExchangeMessage() {
        return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
    }
 
    //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
    // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
    @Bean
    Binding bindingExchangeMessage2() {
        return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
    }
 
}

然后把rabbitmq-provider，rabbitmq-consumer两个项目都跑起来，先调用/sendTopicMessage1 接口：

然后看消费者rabbitmq-consumer的控制台输出情况：
TopicManReceiver监听队列1，绑定键为：topic.man
TopicTotalReceiver监听队列2，绑定键为：topic.#
而当前推送的消息，携带的路由键为：topic.man

所以可以看到两个监听消费者receiver都成功消费到了消息，因为这两个recevier监听的队列的绑定键都能与这条消息携带的路由键匹配上。

接下来调用接口/sendTopicMessage2:

然后看消费者rabbitmq-consumer的控制台输出情况：
TopicManReceiver监听队列1，绑定键为：topic.man
TopicTotalReceiver监听队列2，绑定键为：topic.#
而当前推送的消息，携带的路由键为：topic.woman

所以可以看到两个监听消费者只有TopicTotalReceiver成功消费到了消息。

1.6.3 扇型交换机（Fanout Exchang）

接下来是使用Fanout Exchang 扇型交换机。

同样地，先在rabbitmq-provider项目上创建FanoutRabbitConfig.java：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class FanoutRabbitConfig {
 
    /**
     *  创建三个队列 ：fanout.A   fanout.B  fanout.C
     *  将三个队列都绑定在交换机 fanoutExchange 上
     *  因为是扇型交换机, 路由键无需配置,配置也不起作用
     */
 
    @Bean
    public Queue queueA() {
        return new Queue("fanout.A");
    }
 
    @Bean
    public Queue queueB() {
        return new Queue("fanout.B");
    }
 
    @Bean
    public Queue queueC() {
        return new Queue("fanout.C");
    }
 
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }
 
    @Bean
    Binding bindingExchangeA() {
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeB() {
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeC() {
        return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }
}

然后是写一个接口用于推送消息

    @GetMapping("/sendFanoutMessage")
    public String sendFanoutMessage() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: testFanoutMessage ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("fanoutExchange", null, map);
        return "ok";
    }

接着在rabbitmq-consumer项目里加上消息消费类

FanoutReceiverA.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.***ponent;
import java.util.Map;

@***ponent
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverA消费者收到消息  : " +testMessage.toString());
    }
}

FanoutReceiverB.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.***ponent;
import java.util.Map;

@***ponent
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverB消费者收到消息  : " +testMessage.toString());
    }
}

FanoutReceiverC.java：

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.***ponent;
import java.util.Map;

@***ponent
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {
 
    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消费者收到消息  : " +testMessage.toString());
    }
}

然后加上扇型交换机的配置类，FanoutRabbitConfig.java（消费者真的要加这个配置吗？ 不需要的其实，理由在前面已经说过了）：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutRabbitConfig {
 
    /**
     *  创建三个队列 ：fanout.A   fanout.B  fanout.C
     *  将三个队列都绑定在交换机 fanoutExchange 上
     *  因为是扇型交换机, 路由键无需配置,配置也不起作用
     */
 
 
    @Bean
    public Queue queueA() {
        return new Queue("fanout.A");
    }
 
    @Bean
    public Queue queueB() {
        return new Queue("fanout.B");
    }
 
    @Bean
    public Queue queueC() {
        return new Queue("fanout.C");
    }
 
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }
 
    @Bean
    Binding bindingExchangeA() {
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeB() {
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }
 
    @Bean
    Binding bindingExchangeC() {
        return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }
}

最后将rabbitmq-provider和rabbitmq-consumer项目都跑起来，调用下接口/sendFanoutMessage ：

可以看到只要发送到 fanoutExchange 这个扇型交换机的消息， 三个队列都绑定这个交换机，所以三个消息接收类都监听到了这条消息。

1.6.4 消息回调

到了这里其实三个常用的交换机的使用我们已经完毕了，那么接下来我们继续讲讲消息的回调，其实就是消息确认（生产者推送消息成功，消费者接收消息成功）

1.6.4.1 生产者消息确认回调机制

在rabbitmq-provider项目的application.yml文件上，加上消息确认的配置项后：

ps： 本篇文章使用springboot版本为2.1.7.RELEASE;
如果你们在配置确认回调，测试发现无法触发回调函数，那么存在原因也许是因为版本导致的配置项不起效，可以把publisher-confirms: true 替换为 publisher-confirm-type: correlated

server:
  port: 8021
spring:
  #给项目来个名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 服务器
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    #虚拟host 可以不设置,使用server默认host
    virtual-host: J******Host
    #消息确认配置项
 
    #确认消息已发送到交换机(Exchange)
    publisher-confirms: true
    #确认消息已发送到队列(Queue)
    publisher-returns: true

然后是配置相关的消息确认回调函数，RabbitConfig.java：

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {
 
    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);
 
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("ConfirmCallback:     "+"相关数据："+correlationData);
                System.out.println("ConfirmCallback:     "+"确认情况："+ack);
                System.out.println("ConfirmCallback:     "+"原因："+cause);
            }
        });
 
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("ReturnCallback:     "+"消息："+message);
                System.out.println("ReturnCallback:     "+"回应码："+replyCode);
                System.out.println("ReturnCallback:     "+"回应信息："+replyText);
                System.out.println("ReturnCallback:     "+"交换机："+exchange);
                System.out.println("ReturnCallback:     "+"路由键："+routingKey);
            }
        });
        return rabbitTemplate;
    }
}

到这里，生产者推送消息的消息确认调用回调函数已经完毕。
可以看到上面写了两个回调函数， ConfirmCallback 和RetrunCallback；
那么以上这两种回调函数都是在什么情况会触发呢？

先从总体的情况分析，推送消息存在四种情况：
1、消息推送到server，但是在server里找不到交换机
2、消息推送到server，找到交换机了，但是没找到队列
3、消息推送到sever，交换机和队列啥都没找到
4、消息推送成功

那么我先写几个接口来分别测试和认证下以上4种情况，消息确认触发回调函数的情况：

1、消息推送到server，但是在server里找不到交换机
写个测试接口，把消息推送到名为non-existent-exchange的交换机上（这个交换机是没有创建没有配置的）：

    @GetMapping("/TestMessageAck")
    public String TestMessageAck() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: non-existent-exchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("non-existent-exchange", "TestDirectRouting", map);
        return "ok";
    }

调用接口，查看rabbitmq-provuder项目的控制台输出情况（原因里面有说，没有找到交换机non-existent-exchange）：

2019-09-04 09:37:45.197 ERROR 8172 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'J******Host', class-id=60, method-id=40)
ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：false
ConfirmCallback:     原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'J******Host', class-id=60, method-id=40)

结论： 这种情况触发的是 ConfirmCallback 回调函数。

2、消息推送到server，找到交换机了，但是没找到队列
这种情况就是需要新增一个交换机，但是不给这个交换机绑定队列，我来简单地在DirectRabitConfig里面新增一个直连交换机，名叫lonelyDirectExchange，但没给它做任何绑定配置操作：

    @Bean
    DirectExchange lonelyDirectExchange() {
        return new DirectExchange("lonelyDirectExchange");
    }

然后写个测试接口，把消息推送到名为lonelyDirectExchange的交换机上（这个交换机是没有任何队列配置的）：

    @GetMapping("/TestMessageAck2")
    public String TestMessageAck2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: lonelyDirectExchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("lonelyDirectExchange", "TestDirectRouting", map);
        return "ok";
    }

调用接口，查看rabbitmq-provuder项目的控制台输出情况：

ReturnCallback:     消息：(Body:'{createTime=2019-09-04 09:48:01, messageId=563077d9-0a77-4c27-8794-ecfb183eac80, messageData=message: lonelyDirectExchange test message }' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
ReturnCallback:     回应码：312
ReturnCallback:     回应信息：NO_ROUTE
ReturnCallback:     交换机：lonelyDirectExchange
ReturnCallback:     路由键：TestDirectRouting
ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：true
ConfirmCallback:     原因：null

可以看到这种情况，两个函数都被调用了；
这种情况下，消息是推送成功到服务器了的，所以ConfirmCallback对消息确认情况是true；
而在RetrunCallback回调函数的打印参数里面可以看到，消息是推送到了交换机成功了，但是在路由分发给队列的时候，找不到队列，所以报了错误 NO_ROUTE 。
结论：这种情况触发的是 ConfirmCallback和RetrunCallback两个回调函数。

3、消息推送到sever，交换机和队列啥都没找到
这种情况其实一看就觉得跟第一种很像，没错 ，第三种和第一种情况回调是一致的，所以不做结果说明了。
结论： 这种情况触发的是ConfirmCallback回调函数。

4、消息推送成功
那么测试下，按照正常调用之前消息推送的接口就行，就调用下 /sendFanoutMessage接口，可以看到控制台输出：

ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：true
ConfirmCallback:     原因：null

结论： 这种情况触发的是ConfirmCallback回调函数。

以上是生产者推送消息的消息确认 回调函数的使用介绍（可以在回调函数根据需求做对应的扩展或者业务数据处理）。

1.6.4.2 消费者消息确认机制

和生产者的消息确认机制不同，因为消息接收本来就是在监听消息，符合条件的消息就会消费下来。

所以，消息接收的确认机制主要存在三种模式：
1、自动确认， 这也是默认的消息确认情况。 AcknowledgeMode.NONE
RabbitMQ成功将消息发出（即将消息成功写入TCP Socket）中立即认为本次投递已经被正确处理，不管消费者端是否成功处理本次投递。
所以这种情况如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息。
一般这种情况我们都是使用try catch捕捉异常后，打印日志用于追踪数据，这样找出对应数据再做后续处理。

2、根据情况确认， 这个不做介绍

3、手动确认 ， 这个比较关键，也是我们配置接收消息确认机制时，多数选择的模式。
消费者收到消息后，手动调用basic.ack/basic.nack/basic.reject后，RabbitMQ收到这些消息后，才认为本次投递成功。
basic.ack用于肯定确认
basic.nack用于否定确认（注意：这是AMQP 0-9-1的RabbitMQ扩展）
basic.reject用于否定确认，但与basic.nack相比有一个限制：一次只能拒绝单条消息

消费者端以上的3``个方法都表示消息已经被正确投递，但是basic.ack表示消息已经被正确处理。
而basic.nack,basic.reject表示没有被正确处理：

着重讲下reject，因为有时候一些场景是需要重新入列的。

channel.basicReject(deliveryTag, true); 拒绝消费当前消息，如果第二参数传入true，就是将数据重新丢回队列里，那么下次还会消费这消息。设置false，就是告诉服务器，我已经知道这条消息数据了，因为一些原因拒绝它，而且服务器把这个消息丢掉就行，下次不想再消费这条消息了。

使用拒绝后重新入列这个确认模式要谨慎，因为一般都是出现异常的时候，catch异常再拒绝入列，选择是否重入列。

但是如果使用不当会导致一些每次都被你重入列的消息一直消费-入列-消费-入列这样循环，会导致消息积压。

顺便也简单讲讲 nack，这个也是相当于设置不消费某条消息。

channel.basi***ack(deliveryTag, false, true);

第一个参数依然是当前消息到的数据的唯一id;
第二个参数是指是否针对多条消息；如果是true，也就是说一次性针对当前通道的消息的tagID小于当前这条消息的，都拒绝确认。
第三个参数是指是否重新入列，也就是指不确认的消息是否重新丢回到队列里面去。

同样使用不确认后重新入列这个确认模式要谨慎，因为这里也可能因为考虑不周出现消息一直被重新丢回去的情况，导致积压。

看了上面这么多介绍，接下来我们一起配置下，看看一般的消息接收手动确认是怎么样的。
​​​​​​
在消费者项目里，新建MessageListenerConfig.java上添加代码相关的配置代码：

import ***.elegant.rabbitmqconsumer.receiver.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MessageListenerConfig {
 
    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    private MyAckReceiver myAckReceiver;//消息接收处理类
 
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        // RabbitMQ默认是自动确认，这里改为手动确认消息
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); 
        //设置一个队列
        container.setQueueNames("TestDirectQueue");
        //如果同时设置多个如下： 前提是队列都是必须已经创建存在的
        //  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
 
        //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues
        //container.setQueues(new Queue("TestDirectQueue",true));
        //container.addQueues(new Queue("TestDirectQueue2",true));
        //container.addQueues(new Queue("TestDirectQueue3",true));
        container.setMessageListener(myAckReceiver);
 
        return container;
    }
}

对应的手动确认消息监听类，MyAckReceiver.java（手动确认模式需要实现 ChannelAwareMessageListener）：
//之前的相关监听器可以先注释掉，以免造成多个同类型监听器都监听同一个队列。
//这里的获取消息转换，只作参考，如果报数组越界可以自己根据格式去调整。

import ***.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.***ponent;
import java.util.HashMap;
import java.util.Map;
 
@***ponent
 
public class MyAckReceiver implements ChannelAwareMessageListener {
 
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因为传递消息的时候用的map传递,所以将Map从Message内取出需要做些处理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以点进Message里面看源码,单引号直接的数据就是我们的map消息数据
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3);
            String messageId=msgMap.get("messageId");
            String messageData=msgMap.get("messageData");
            String createTime=msgMap.get("createTime");
            System.out.println("  MyAckReceiver  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
            System.out.println("消费的主题消息来自："+message.getMessageProperties().getConsumerQueue());
            channel.basicAck(deliveryTag, true); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
//			channel.basicReject(deliveryTag, true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
 
     //{key=value,key=value,key=value} 格式转换成map
    private Map<String, String> mapStringToMap(String str,int entryNum ) {
        str = str.substring(1, str.length() - 1);
        String[] strs = str.split(",",entryNum);
        Map<String, String> map = new HashMap<String, String>();
        for (String string : strs) {
            String key = string.split("=")[0].trim();
            String value = string.split("=")[1];
            map.put(key, value);
        }
        return map;
    }
}

这时，先调用接口/sendDirectMessage， 给直连交换机TestDirectExchange 的队列TestDirectQueue 推送一条消息，可以看到监听器正常消费了下来：

到这里，我们其实已经掌握了怎么去使用消息消费的手动确认了。
但是这个场景往往不够！ 因为很多伙伴之前给我评论反应，他们需要这个消费者项目里面，监听的好几个队列都想变成手动确认模式，而且处理的消息业务逻辑不一样。

没有问题，接下来看代码

场景： 除了直连交换机的队列TestDirectQueue需要变成手动确认以外，我们还需要将一个其他的队列

或者多个队列也变成手动确认，而且不同队列实现不同的业务处理。

那么我们需要做的第一步，往SimpleMessageListenerContainer里添加多个队列：

然后我们的手动确认消息监听类，MyAckReceiver.java 就可以同时将上面设置到的队列的消息都消费下来。

但是我们需要做不用的业务逻辑处理，那么只需要 根据消息来自的队列名进行区分处理即可，如：

import ***.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.***ponent;
import java.util.HashMap;
import java.util.Map;
 
@***ponent
public class MyAckReceiver implements ChannelAwareMessageListener {
 
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因为传递消息的时候用的map传递,所以将Map从Message内取出需要做些处理
            String msg = message.toString();
            String[] msgArray = msg.split("'");//可以点进Message里面看源码,单引号直接的数据就是我们的map消息数据
            Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3);
            String messageId=msgMap.get("messageId");
            String messageData=msgMap.get("messageData");
            String createTime=msgMap.get("createTime");
            
            if ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消费到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                System.out.println("执行TestDirectQueue中的消息的业务处理流程......");  
            }
 
            if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue());
                System.out.println("消息成功消费到  messageId:"+messageId+"  messageData:"+messageData+"  createTime:"+createTime);
                System.out.println("执行fanout.A中的消息的业务处理流程......");
 
            }
            
            channel.basicAck(deliveryTag, true);
//			channel.basicReject(deliveryTag, true);//为true会重新放回队列
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
 
    //{key=value,key=value,key=value} 格式转换成map
    private Map<String, String> mapStringToMap(String str,int enNum) {
        str = str.substring(1, str.length() - 1);
        String[] strs = str.split(",",enNum);
        Map<String, String> map = new HashMap<String, String>();
        for (String string : strs) {
            String key = string.split("=")[0].trim();
            String value = string.split("=")[1];
            map.put(key, value);
        }
        return map;
    }
}

ok，这时候我们来分别往不同队列推送消息，看看效果：

调用接口/sendDirectMessage 和/sendFanoutMessage，

如果你还想新增其他的监听队列，也就是按照这种方式新增配置即可（或者完全可以分开多个消费者项目去监听处理）

转载请说明出处内容投诉
CSS教程_站长资源网 » RabbitMQ详解，用心看完这一篇就够了【重点】
翔宽

翔宽

分享到：

相关推荐

发表评论

欢迎 访客 发表评论

一个令你着迷的主题！

查看演示 官网购买