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

1.1 消息中间件

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

1.1.1 应用场景

1.1.1.1 异步处理

场景说明

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

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

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

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

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

RabbitMQ是由erlang语言开发基于AMQPAdvanced 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 –import https://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的概要信息 , 如消息的数量ConnectionChannelExchangeQueueConsumer的数量

1.4.1.6.2 Exchange 交换器

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

上面的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-consumer消息没有被消费的我们去rabbitMq管理页面看看是否推送成功

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

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

然后看消费者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.Component;
import java.util.Map;

@Component
@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.Component;
import java.util.Map;

@Component
@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.Component;
import java.util.Map;

@Component
@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: JCcccHost
    #消息确认配置项
 
    #确认消息已发送到交换机(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 'JCcccHost', 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 'JCcccHost', 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.basicNack(deliveryTag, false, true);

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

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

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

import com.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 com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
 
@Component
 
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 com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
 
@Component
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

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