0


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

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.com/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 –importhttps://www.rabbitmq.com/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

做成

Window

s服务
以管理员身份运行

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

1.6.1 直连交换机(direct exchange)

接着我们先使用下

direct exchange

(直连型交换机),创建

DirectRabbitConfig.java

(对于队列和交换机持久化以及连接使用设置,在注释里有说明,后面的不同交换机的配置就不做同样说明了):

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

然后写个简单的接口进行消息推送(根据需求也可以改为定时任务等等,具体看需求),

SendMessageController.java

importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.GetMapping;importorg.springframework.web.bind.annotation.RestController;importjava.time.LocalDateTime;importjava.time.format.DateTimeFormatter;importjava.util.HashMap;importjava.util.Map;importjava.util.UUID;@RestControllerpublicclassSendMessageController{//使用RabbitTemplate,这提供了接收/发送等等方法@AutowiredRabbitTemplate rabbitTemplate;@GetMapping("/sendDirectMessage")publicStringsendDirectMessage(){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=newHashMap<>();
        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:JCcccHost

然后一样,创建

DirectRabbitConfig.java

(消费者单纯的使用,其实可以不用添加这个配置,直接建后面的监听就好,使用注解来让监听器监听对应的队列即可。配置上了的话,其实消费者也是生成者的身份,也能推送该消息):

importorg.springframework.amqp.core.Binding;importorg.springframework.amqp.core.BindingBuilder;importorg.springframework.amqp.core.DirectExchange;importorg.springframework.amqp.core.Queue;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassDirectRabbitConfig{//队列 起名:TestDirectQueue@BeanpublicQueueTestDirectQueue(){returnnewQueue("TestDirectQueue",true);}//Direct交换机 起名:TestDirectExchange@BeanDirectExchangeTestDirectExchange(){returnnewDirectExchange("TestDirectExchange");}//绑定  将队列和交换机绑定, 并设置用于匹配键:TestDirectRouting@BeanBindingbindingDirect(){returnBindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");}}

然后是创建消息接收监听类,

DirectReceiver.java

@Component//监听的队列名称 TestDirectQueue@RabbitListener(queues ="TestDirectQueue")publicclassDirectReceiver{@RabbitHandlerpublicvoidprocess(Map testMessage){System.out.println("DirectReceiver消费者收到消息  : "+ testMessage.toString());}}

然后将

rabbitmq-consumer

项目运行起来,可以看到把之前推送的那条消息消费下来了:
在这里插入图片描述
然后可以再继续调用

rabbitmq-provider

项目的推送消息接口,可以看到消费者即时消费消息:
在这里插入图片描述
那么直连交换机既然是一对一,那如果咱们配置多台监听绑定到同一个直连交互的同一个队列,会怎么样?
在这里插入图片描述
可以看到是实现了轮询的方式对消息进行消费,而且不存在重复消费。

1.6.2 主题交换机(Topic Exchange)

接着,我们使用

Topic Exchange

主题交换机。

rabbitmq-provider

项目里面创建

TopicRabbitConfig.java

importorg.springframework.amqp.core.Binding;importorg.springframework.amqp.core.BindingBuilder;importorg.springframework.amqp.core.Queue;importorg.springframework.amqp.core.TopicExchange;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassTopicRabbitConfig{//绑定键publicfinalstaticString man ="topic.man";publicfinalstaticString woman ="topic.woman";@BeanpublicQueuefirstQueue(){returnnewQueue(TopicRabbitConfig.man);}@BeanpublicQueuesecondQueue(){returnnewQueue(TopicRabbitConfig.woman);}@BeanTopicExchangeexchange(){returnnewTopicExchange("topicExchange");}//将firstQueue和topicExchange绑定,而且绑定的键值为topic.man//这样只要是消息携带的路由键是topic.man,才会分发到该队列@BeanBindingbindingExchangeMessage(){returnBindingBuilder.bind(firstQueue()).to(exchange()).with(man);}//将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#// 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列@BeanBindingbindingExchangeMessage2(){returnBindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");}}

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

@GetMapping("/sendTopicMessage1")publicStringsendTopicMessage1(){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 =newHashMap<>();
        manMap.put("messageId", messageId);
        manMap.put("messageData", messageData);
        manMap.put("createTime", createTime);
        rabbitTemplate.convertAndSend("topicExchange","topic.man", manMap);return"ok";}@GetMapping("/sendTopicMessage2")publicStringsendTopicMessage2(){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 =newHashMap<>();
        womanMap.put("messageId", messageId);
        womanMap.put("messageData", messageData);
        womanMap.put("createTime", createTime);
        rabbitTemplate.convertAndSend("topicExchange","topic.woman", womanMap);return"ok";}}

生产者这边已经完事,先不急着运行,在

rabbitmq-consumer

项目上,创建

TopicManReceiver.java

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;importjava.util.Map;@Component@RabbitListener(queues ="topic.man")publicclassTopicManReceiver{@RabbitHandlerpublicvoidprocess(Map testMessage){System.out.println("TopicManReceiver消费者收到消息  : "+ testMessage.toString());}}

再创建一个

TopicTotalReceiver.java

packagecom.elegant.rabbitmqconsumer.receiver;importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;importjava.util.Map;@Component@RabbitListener(queues ="topic.woman")publicclassTopicTotalReceiver{@RabbitHandlerpublicvoidprocess(Map testMessage){System.out.println("TopicTotalReceiver消费者收到消息  : "+ testMessage.toString());}}

同样,加主题交换机的相关配置,

TopicRabbitConfig.java

(消费者一定要加这个配置吗? 不需要的其实,理由在前面已经说过了):

importorg.springframework.amqp.core.Binding;importorg.springframework.amqp.core.BindingBuilder;importorg.springframework.amqp.core.Queue;importorg.springframework.amqp.core.TopicExchange;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassTopicRabbitConfig{//绑定键publicfinalstaticString man ="topic.man";publicfinalstaticString woman ="topic.woman";@BeanpublicQueuefirstQueue(){returnnewQueue(TopicRabbitConfig.man);}@BeanpublicQueuesecondQueue(){returnnewQueue(TopicRabbitConfig.woman);}@BeanTopicExchangeexchange(){returnnewTopicExchange("topicExchange");}//将firstQueue和topicExchange绑定,而且绑定的键值为topic.man//这样只要是消息携带的路由键是topic.man,才会分发到该队列@BeanBindingbindingExchangeMessage(){returnBindingBuilder.bind(firstQueue()).to(exchange()).with(man);}//将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#// 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列@BeanBindingbindingExchangeMessage2(){returnBindingBuilder.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

importorg.springframework.amqp.core.Binding;importorg.springframework.amqp.core.BindingBuilder;importorg.springframework.amqp.core.FanoutExchange;importorg.springframework.amqp.core.Queue;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassFanoutRabbitConfig{/**
     *  创建三个队列 :fanout.A   fanout.B  fanout.C
     *  将三个队列都绑定在交换机 fanoutExchange 上
     *  因为是扇型交换机, 路由键无需配置,配置也不起作用
     */@BeanpublicQueuequeueA(){returnnewQueue("fanout.A");}@BeanpublicQueuequeueB(){returnnewQueue("fanout.B");}@BeanpublicQueuequeueC(){returnnewQueue("fanout.C");}@BeanFanoutExchangefanoutExchange(){returnnewFanoutExchange("fanoutExchange");}@BeanBindingbindingExchangeA(){returnBindingBuilder.bind(queueA()).to(fanoutExchange());}@BeanBindingbindingExchangeB(){returnBindingBuilder.bind(queueB()).to(fanoutExchange());}@BeanBindingbindingExchangeC(){returnBindingBuilder.bind(queueC()).to(fanoutExchange());}}

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

@GetMapping("/sendFanoutMessage")publicStringsendFanoutMessage(){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 =newHashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        rabbitTemplate.convertAndSend("fanoutExchange",null, map);return"ok";}

接着在

rabbitmq-consumer

项目里加上消息消费类

FanoutReceiverA.java

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;importjava.util.Map;@Component@RabbitListener(queues ="fanout.A")publicclassFanoutReceiverA{@RabbitHandlerpublicvoidprocess(Map testMessage){System.out.println("FanoutReceiverA消费者收到消息  : "+testMessage.toString());}}
FanoutReceiverB.java

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;importjava.util.Map;@Component@RabbitListener(queues ="fanout.B")publicclassFanoutReceiverB{@RabbitHandlerpublicvoidprocess(Map testMessage){System.out.println("FanoutReceiverB消费者收到消息  : "+testMessage.toString());}}
FanoutReceiverC.java

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;importjava.util.Map;@Component@RabbitListener(queues ="fanout.C")publicclassFanoutReceiverC{@RabbitHandlerpublicvoidprocess(Map testMessage){System.out.println("FanoutReceiverC消费者收到消息  : "+testMessage.toString());}}

然后加上扇型交换机的配置类,

FanoutRabbitConfig.java

(消费者真的要加这个配置吗? 不需要的其实,理由在前面已经说过了):

importorg.springframework.amqp.core.Binding;importorg.springframework.amqp.core.BindingBuilder;importorg.springframework.amqp.core.FanoutExchange;importorg.springframework.amqp.core.Queue;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassFanoutRabbitConfig{/**
     *  创建三个队列 :fanout.A   fanout.B  fanout.C
     *  将三个队列都绑定在交换机 fanoutExchange 上
     *  因为是扇型交换机, 路由键无需配置,配置也不起作用
     */@BeanpublicQueuequeueA(){returnnewQueue("fanout.A");}@BeanpublicQueuequeueB(){returnnewQueue("fanout.B");}@BeanpublicQueuequeueC(){returnnewQueue("fanout.C");}@BeanFanoutExchangefanoutExchange(){returnnewFanoutExchange("fanoutExchange");}@BeanBindingbindingExchangeA(){returnBindingBuilder.bind(queueA()).to(fanoutExchange());}@BeanBindingbindingExchangeB(){returnBindingBuilder.bind(queueB()).to(fanoutExchange());}@BeanBindingbindingExchangeC(){returnBindingBuilder.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: JCcccHost
    #消息确认配置项
 
    #确认消息已发送到交换机(Exchange)
    publisher-confirms: true
    #确认消息已发送到队列(Queue)
    publisher-returns: true

然后是配置相关的消息确认回调函数,

RabbitConfig.java

importorg.springframework.amqp.core.Message;importorg.springframework.amqp.rabbit.connection.ConnectionFactory;importorg.springframework.amqp.rabbit.connection.CorrelationData;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassRabbitConfig{@BeanpublicRabbitTemplatecreateRabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate =newRabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);//设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);
 
        rabbitTemplate.setConfirmCallback(newRabbitTemplate.ConfirmCallback(){@Overridepublicvoidconfirm(CorrelationData correlationData,boolean ack,String cause){System.out.println("ConfirmCallback:     "+"相关数据:"+correlationData);System.out.println("ConfirmCallback:     "+"确认情况:"+ack);System.out.println("ConfirmCallback:     "+"原因:"+cause);}});
 
        rabbitTemplate.setReturnCallback(newRabbitTemplate.ReturnCallback(){@OverridepublicvoidreturnedMessage(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")publicStringTestMessageAck(){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 =newHashMap<>();
        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-0409: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 'JCcccHost',class-id=60, method-id=40)ConfirmCallback:     相关数据:nullConfirmCallback:     确认情况:falseConfirmCallback:     原因:channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'JCcccHost',class-id=60, method-id=40)

**结论: 这种情况触发的是

ConfirmCallback

回调函数。**

**2、消息推送到

server

,找到交换机了,但是没找到队列**
这种情况就是需要新增一个交换机,但是不给这个交换机绑定队列,我来简单地在

DirectRabitConfig

里面新增一个直连交换机,名叫

lonelyDirectExchange

,但没给它做任何绑定配置操作:

@BeanDirectExchangelonelyDirectExchange(){returnnewDirectExchange("lonelyDirectExchange");}

然后写个测试接口,把消息推送到名为

lonelyDirectExchange

的交换机上(这个交换机是没有任何队列配置的):

@GetMapping("/TestMessageAck2")publicStringTestMessageAck2(){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 =newHashMap<>();
        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:     回应码:312ReturnCallback:     回应信息:NO_ROUTE
ReturnCallback:     交换机:lonelyDirectExchange
ReturnCallback:     路由键:TestDirectRoutingConfirmCallback:     相关数据:nullConfirmCallback:     确认情况:trueConfirmCallback:     原因:null

可以看到这种情况,两个函数都被调用了;
这种情况下,消息是推送成功到服务器了的,所以

ConfirmCallback

对消息确认情况是

true


而在

RetrunCallback

回调函数的打印参数里面可以看到,消息是推送到了交换机成功了,但是在路由分发给队列的时候,找不到队列,所以报了错误

NO_ROUTE


**结论:这种情况触发的是

ConfirmCallback

RetrunCallback

两个回调函数**。

**3、消息推送到

sever

,交换机和队列啥都没找到**
这种情况其实一看就觉得跟第一种很像,没错 ,第三种和第一种情况回调是一致的,所以不做结果说明了。
**结论: 这种情况触发的是

ConfirmCallback

回调函数**。

4、消息推送成功
那么测试下,按照正常调用之前消息推送的接口就行,就调用下

/sendFanoutMessage

接口,可以看到控制台输出:

ConfirmCallback:     相关数据:nullConfirmCallback:     确认情况:trueConfirmCallback:     原因: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.basicNack(deliveryTag,false,true);

第一个参数依然是当前消息到的数据的唯一

id

;
第二个参数是指是否针对多条消息;如果是true,也就是说一次性针对当前通道的消息的

tagID

小于当前这条消息的,都拒绝确认。
第三个参数是指是否重新入列,也就是指不确认的消息是否重新丢回到队列里面去。

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

看了上面这么多介绍,接下来我们一起配置下,看看一般的消息接收手动确认是怎么样的。
​​​​​​
在消费者项目里,新建

MessageListenerConfig.java

上添加代码相关的配置代码:

importcom.elegant.rabbitmqconsumer.receiver.MyAckReceiver;importorg.springframework.amqp.core.AcknowledgeMode;importorg.springframework.amqp.core.Queue;importorg.springframework.amqp.rabbit.connection.CachingConnectionFactory;importorg.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassMessageListenerConfig{@AutowiredprivateCachingConnectionFactory connectionFactory;@AutowiredprivateMyAckReceiver myAckReceiver;//消息接收处理类@BeanpublicSimpleMessageListenerContainersimpleMessageListenerContainer(){SimpleMessageListenerContainer container =newSimpleMessageListenerContainer(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

):
//之前的相关监听器可以先注释掉,以免造成多个同类型监听器都监听同一个队列。
//这里的获取消息转换,只作参考,如果报数组越界可以自己根据格式去调整。

importcom.rabbitmq.client.Channel;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;importorg.springframework.stereotype.Component;importjava.util.HashMap;importjava.util.Map;@ComponentpublicclassMyAckReceiverimplementsChannelAwareMessageListener{@OverridepublicvoidonMessage(Message message,Channel channel)throwsException{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} 格式转换成mapprivateMap<String,String>mapStringToMap(String str,int entryNum ){
        str = str.substring(1, str.length()-1);String[] strs = str.split(",",entryNum);Map<String,String> map =newHashMap<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

就可以同时将上面设置到的队列的消息都消费下来。

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

importcom.rabbitmq.client.Channel;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;importorg.springframework.stereotype.Component;importjava.util.HashMap;importjava.util.Map;@ComponentpublicclassMyAckReceiverimplementsChannelAwareMessageListener{@OverridepublicvoidonMessage(Message message,Channel channel)throwsException{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} 格式转换成mapprivateMap<String,String>mapStringToMap(String str,int enNum){
        str = str.substring(1, str.length()-1);String[] strs = str.split(",",enNum);Map<String,String> map =newHashMap<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


在这里插入图片描述
如果你还想新增其他的监听队列,也就是按照这种方式新增配置即可(或者完全可以分开多个消费者项目去监听处理)


本文转载自: https://blog.csdn.net/weixin_42039228/article/details/123493937
版权归原作者 sunnyday0426 所有, 如有侵权,请联系我们删除。

“RabbitMQ详解,用心看完这一篇就够了【重点】”的评论:

还没有评论