一文玩转RabbitMQ

#消息队列 #MQ

1. MQ概述

1.1 MQ简介

MQ 全称为Message Queue, 消息队列。是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。多用于分布式系统之间进行通信。

消息传递指的是程序之间通过在消息中间件中发送数据进行通信，而不是通过直接调用彼此来通信。队列的使用除去了接收和发送应用程序同时执行的要求，实现了应用解偶。

在项目中，通常会将一些无需即时返回且耗时的操作提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。

1.2 MQ的优势

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

1.3 MQ的劣势

系统可用性降低

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

系统复杂度提高

MQ 的加入大大增加了系统的复杂度,以前系统间是同步的远程调用,现在是通过 MQ 进行异步调用。如何保证消息没有被重复消费?怎么处理消息丢失情况?那么保证消息传递的顺序性?

一致性问题

A 系统处理完业务,通过 MQ 给B、C、D三个系统发消息数据,如果 B 系统、C 系统处理成功,D 系统处理
失败。如何保证消息数据处理的一致性?

1.4 常见的MQ产品

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

2. RabbitMQ概述

2.1 AMQP协议

AMQP，即Advanced Message Queuing Protocol(高级消息队列协议)，一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端中间件不同产品，不同的开发语言等条件的限制。类似于http、ftp这些协议。

2.2 RabbitMQ

RabbitMQ是一个在AMQP（Advanced Message Queuing Protocol ）基础上实现的，可复用的企业消息系统。它可以用于大型软件系统各个模块之间的高效通信，支持高并发，支持可扩展。

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

2.3 RabbitMQ结构图

2.4 核心概念

Broker（重要）：简单来说就是消息队列服务器实体。
Exchange（重要）：消息交换机，它指定消息按什么规则，路由到哪个队列。
Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
Binding：绑定，它的作用就是把exchange通过RoutingKey和queue绑定起来。
Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
Virtual Host：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务

2.5 RabbitMQ工作流程

消息在Producer中产生
Producer连接到RabbitMQ服务端，创建一个连接Channel。
Producer声明Exchange，并设置相关属性。
Producer声明Queue，并设置属性。
Producer通过RoutingKey将Exchange和Queue绑定。
Producer将消息投递到Exchange。
Exchange根据绑定规则进行消息路由，将消息投递到符合规范的Queue中。

3. RabbitMQ安装

ubuntu安装rabbitmq_风吹过的烟花的技术博客_51CTO博客
Centos-RabbitMQ安装说明文档

4. Rabbit工作模式

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

官方文档:https://www.rabbitmq.com/getstarted.html

该文档使用Java语言作为讲解使用，涉及到的技术有： JMS。
JMS 即 Java 消息服务(JavaMessage Service)应用程序接口,是一个 Java 平台中关于面向消息中间件的API，类似于JDBC。

4.1 HelloWorld-简单模式

简单模式中，只包含一个生产者和消费者以及一个消息队列。注意，这种情况下queue的名称要和routingKey的值保持一一致。简单模式下不需要设置交换机，它会自动使用默认的名为""的交换机。

生产者代码：

创建连接工厂
设置参数
创建Connection对象
通过Connection对象创建channel
通过channel声明queue
发布数据到queue。
释放资源

public static void main(String[] args) throws IOException, TimeoutException {  // 1. 创建连接工厂  ConnectionFactory connectionFactory = new ConnectionFactory();  // 2. 设置参数  connectionFactory.setHost("22.22.22.22"); // ip 默认为127.0.0.1  connectionFactory.setPort(5672);            // 端口 默认为5672  connectionFactory.setVirtualHost("/");  connectionFactory.setUsername("admin");  connectionFactory.setPassword("123123");  // 3. 创建Connection  Connection connection = connectionFactory.newConnection();  // 4. 创建channel  Channel channel = connection.createChannel();  // 5. 声明queue，保存数据  /*  queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments    1.queue: 队列名称  2.durable: 是否持久化，当MQ重启后，数据仍在。 3.exclusive：  - 是否独占。即只允许一个消费者监听当前队列。  - 当connection关闭时，是否删除当前队列。  4.autoDelete: 是否自动删除。当没有消费者时，自动删除队列。  5.argument：其它参数。  */  channel.queueDeclare("hello_world",true,false,false,null);  // 6. 发送消息到queue  /*    basicPublish(String exchange, String routingKey,  BasicProperties props, byte[] body)       exchange: 交换机的名称,简单模式下会使用默认的名为‘’的交换机。  routingKey： 路由名称，简单模式下使用默认的路由，传值为“”即可。props： 配置信息  body: 需要发送保存的数据  */    String data= "hello rabbiMQ";  channel.basicPublish("","hello_world",null,data.getBytes());  // 7. 释放资源  channel.close();  connection.close();
}

消费端代码：

创建连接工厂
设置参数
创建Connection对象
通过Connection对象创建channel
通过channel声明queue
通过cutomer消费消息。

public static void main(String[] args) throws IOException, TimeoutException {  // 1. 创建连接工厂 ConnectionFactory    ConnectionFactory connectionFactory = new ConnectionFactory();  // 2. 配置信息  connectionFactory.setHost("222.222.222.22");  connectionFactory.setVirtualHost("/");  connectionFactory.setPort(5672);  connectionFactory.setUsername("admin");  connectionFactory.setPassword("123123");  // 3.  创建连接   Connection    Connection connection = connectionFactory.newConnection();  // 4. 创建channel  Channel channel = connection.createChannel();  // 5.  声明 并获取队列"  channel.queueDeclare("hello_rabbitmq",true,false,false,null);  // 6. 获取消息 匿名类 Consumer consumer = new DefaultConsumer(channel){  /**  * 回调方法 当customer 接受到消息后自动执行该部分代码  * @param consumerTag 标识  * @param envelope 一些常用信息，包含路由、交换机等  * @param properties 配置信息  * @param body  数据  * @throws IOException  */  @Override  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  System.out.println("consumerTag:"+consumerTag);  System.out.println("Exchange:"+envelope.getExchange());  System.out.println("RoutingKey(:"+envelope.getRoutingKey());  System.out.println("properties:"+properties);  System.out.println("body:"+new String(body));  }  };  /*  String basicConsume(String queue, boolean autoAck, Consumer callback)        queue: 队列名称  autoAck: 是否自动确认  callback: 回调函数  */    channel.basicConsume("hello_rabbitmq",true,consumer);
}

4.2 WorkQueues模式

与简单模式相比,多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。多个消费者之间对于同一个消息的关系是竞争的关系，若其中一个消费端消费了一个消息，则其它消费端则不会再次消费该消息。该模式主要用于任务过重或任务较多情况下，使用工作队列可以提高任务处理的速度。
代码和简单模式基本一样，只是多了一些消费端，此处不再赘述。

4.3 PubSub-订阅模式

在订阅模型中,多了一个 Exchange 角色,而且过程略有变化，Produce不再将消息直接发布到队列中，而是将消息发布给交换机，交换机根据Routing Key规则将消息传递给队列，这个过程中，交换机是不存储消息的。
其中交换机的规则有以下几种：

➢ Fanout:广播,将消息交给所有绑定到交换机的队列，也是这一小节的主要内容。
➢ Direct:定向,把消息交给符合指定routing key 的队列，即路由模式。
➢ Topic:通配符,把消息交给符合routing pattern(路由模式) 的队列，即通配符模式。
发布端需要声明交换机，且将队列和交换机绑定。

订阅模式下的广播模式，只有一个发布者，多个订阅者，且多个订阅者之间属于同等地位，消费同一个消息。


public static void main(String[] args) throws IOException, TimeoutException {  // 1. 创建连接工厂  ConnectionFactory connectionFactory = new ConnectionFactory();  // 2. 配置参数  connectionFactory.setHost("222.222.222.22"); // ip 默认为127.0.0.1  connectionFactory.setPort(5672);            // 端口 默认为5672  connectionFactory.setVirtualHost("/");  connectionFactory.setUsername("admin");  connectionFactory.setPassword("123123");  // 3. 创建连接  Connection connection = connectionFactory.newConnection();  // 3. 创建 channel        Channel channel = connection.createChannel();  // 4. 创建交换机  /*  exchangeDeclare(String exchange,: 交换机名称  BuiltinExchangeType type,        ：交换机类型 通过BuiltinExchangeType获取  boolean durable                  ：是否持久化  boolean autoDelete,              ： 是否自动删除  boolean internal,                ：是否仅mq内部使用  Map<String, Object> arguments)   ： 参数  */        String exchangeName = "pubSub_test";  channel.exchangeDeclare(exchangeName,BuiltinExchangeType.FANOUT,true,false,false,null);  // 5. 创建队列  String queue1Name = "info_test";  String queue2Name = "error_test";  channel.queueDeclare(queue1Name,true,false,false,null);  channel.queueDeclare(queue2Name,true,false,false,null);  // 6. 绑定交换机和队列  /*  queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments)       queue       : 队列名称  exchange    : 交换机名称  routingKey  : 路由规则，广播模式（fanout）的规则默认为 “”   arguments   : 参数  */        channel.queueBind(queue1Name,exchangeName,"");  channel.queueBind(queue2Name,exchangeName,"");  // 7. 发布消息  String data = "pubSub订阅模式";  channel.basicPublish(exchangeName,"",null,data.getBytes());  // 8. 释放资源  channel.close();  connection.close();
}

订阅端只需绑定queue即可，不同的订阅端可以同一消息进行不同操作，这里仅提供一个订阅端代码。

public static void main(String[] args) throws IOException, TimeoutException {  // 1. 创建连接工厂 ConnectionFactory    ConnectionFactory connectionFactory = new ConnectionFactory();  // 2. 配置信息  connectionFactory.setHost("22.222.222.22");  connectionFactory.setVirtualHost("/");  connectionFactory.setPort(5672);  connectionFactory.setUsername("admin");  connectionFactory.setPassword("123123");  // 3.  创建连接   Connection    Connection connection = connectionFactory.newConnection();  // 4. 创建channel  Channel channel = connection.createChannel();  // 5.  声明 并获取队列"  channel.queueDeclare("info",true,false,false,null);  // 6. 获取消息  Consumer consumer = new DefaultConsumer(channel){  /**  * 回调方法 当接受到消息后自动执行该部分代码  * @param consumerTag 标识  * @param envelope 一些常用信息，包含路由、交换机等  * @param properties 配置信息  * @param body  数据  * @throws IOException  */  @Override  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  System.out.println("模拟输出到控制台操作");  System.out.println("body:"+new String(body));  }  };  /*  String basicConsume(String queue, boolean autoAck, Consumer callback)        queue: 队列名称  autoAck: 自动确认  callback: 回调函数  */    channel.basicConsume("info",true,consumer);
}

发布订阅模式与工作队列模式的区别:

工作队列模式不用定义交换机,而发布/订阅模式需要定义交换机
发布/订阅模式的生产方是面向交换机发送消息,工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)
发布/订阅模式需要设置队列和交换机的绑定,工作队列模式不需要设置,实际上工作队列模式会将队列绑定到默认的交换机。

4.4 Routing-路由模式

Routing 模式要求队列在绑定交换机时要指定 routing key,exchange只会将消息会转发到符合 routing key 的队列上。此时的交换机模式为:Direct。

public static void main(String[] args) throws IOException, TimeoutException {  // 1. 创建连接工厂  ConnectionFactory connectionFactory = new ConnectionFactory();  // 2. 配置参数  connectionFactory.setHost("123.123.123.123"); // ip 默认为127.0.0.1  connectionFactory.setPort(5672);            // 端口 默认为5672  connectionFactory.setVirtualHost("/");  connectionFactory.setUsername("admin");  connectionFactory.setPassword("121232");  // 3. 创建连接  Connection connection = connectionFactory.newConnection();  // 3. 创建 channel        Channel channel = connection.createChannel();  // 4. 创建交换机  /*  exchangeDeclare(String exchange,: 交换机名称  BuiltinExchangeType type,        ：交换机类型 通过BuiltinExchangeType获取  boolean durable                  ：是否持久化  boolean autoDelete,              ： 是否自动删除  boolean internal,                ：是否仅mq内部使用  Map<String, Object> arguments)   ： 参数  */        String exchangeName = "routing_test";  channel.exchangeDeclare(exchangeName,BuiltinExchangeType.DIRECT,true,false,false,null);  // 5. 创建队列  String queue1Name = "queue1";  String queue2Name = "queue2";  channel.queueDeclare(queue1Name,true,false,false,null);  channel.queueDeclare(queue2Name,true,false,false,null);  // 6. 绑定交换机和队列  /*  queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments)     queue       : 队列名称  exchange    : 交换机名称  routingKey  : 路由规则，广播模式（fanout）的规则默认为 “”            arguments   : 参数  */        channel.queueBind(queue1Name,exchangeName,"info");  channel.queueBind(queue1Name,exchangeName,"warning");  channel.queueBind(queue1Name,exchangeName,"error");  channel.queueBind(queue2Name,exchangeName,"error");  // 7. 发布消息  String body = "routing路由模式—---info";  channel.basicPublish(exchangeName,"info",null,body.getBytes());  // 8. 释放资源  channel.close();  connection.close();
}

生产端通过对不同的队列绑定不同的routing key来实现路由功能，消费端只需绑定对应的queue即可。

4.5 Topic-通配符/主题模式

Topic 类型与 Direct 相比,都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型
Exchange 可以让队列在绑定 Routing key 的时候使用通配符!

Routingkey 一般都是由一个或多个单词组成,多个单词之间以”.”分割,例如: item.insert
通配符规则:

通配符	作用	举例
`#`	匹配一个或多个单词，即多级定义	item.# 能够匹配 item.insert.abc
`*`	仅能匹配1个单词，即子级定义	item.* 只能匹配 item.insert

public static void main(String[] args) throws IOException, TimeoutException {  // 1. 创建连接工厂  ConnectionFactory connectionFactory = new ConnectionFactory();  // 2. 配置参数  connectionFactory.setHost("12.11.11.11"); // ip 默认为127.0.0.1  connectionFactory.setPort(5672);            // 端口 默认为5672  connectionFactory.setVirtualHost("/");  connectionFactory.setUsername("admin");  connectionFactory.setPassword("123123****");  // 3. 创建连接  Connection connection = connectionFactory.newConnection();  // 3. 创建 channel        Channel channel = connection.createChannel();  // 4. 创建交换机  /*  exchangeDeclare(String exchange,: 交换机名称  BuiltinExchangeType type,        ：交换机类型 通过BuiltinExchangeType获取  boolean durable                  ：是否持久化  boolean autoDelete,              ： 是否自动删除  boolean internal,                ：是否仅mq内部使用  Map<String, Object> arguments)   ： 参数  */        String exchangeName = "topic_test";  channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);  // 5. 创建队列  String queue1Name = "queue_topic_1";  String queue2Name = "queue_topic_2";  channel.queueDeclare(queue1Name,true,false,false,null);  channel.queueDeclare(queue2Name,true,false,false,null);  // 6. 绑定交换机和队列  /*  queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments)   queue       : 队列名称  exchange    : 交换机名称  routingKey  : 路由规则，广播模式（fanout）的规则默认为 “”            arguments   : 参数  */        channel.queueBind(queue1Name,exchangeName,"log.#");  channel.queueBind(queue1Name,exchangeName,"#.error.#");  channel.queueBind(queue1Name,exchangeName,"#.#");  channel.queueBind(queue2Name,exchangeName,"#.error.#");  // 7. 发布消息
//        String body = "routing路由模式—---info";  String body = "topic通配符模式";  channel.basicPublish(exchangeName,"log.info",null,body.getBytes());  // 8. 释放资源  channel.close();  connection.close();
}

4.6 工作模式小结

简单模式 HelloWorld
一个生产者、一个消费者,不需要设置交换机(使用默认的交换机)。
工作队列模式 Work Queue
一个生产者、多个消费者(竞争关系),不需要设置交换机(使用默认的交换机)。
发布订阅模式 Publish/subscribe
需要设置类型为 fanout 的交换机,并且交换机和队列进行绑定,当发送消息到交换机后,交换机会将消
息发送到绑定的队列。
路由模式 Routing
需要设置类型为 direct 的交换机,交换机和队列进行绑定,并且指定 routing key,当发送消息到交换机
后,交换机会根据 routing key 将消息发送到对应的队列。
通配符模式 Topic
需要设置类型为 topic 的交换机,交换机和队列进行绑定,并且指定通配符方式的 routing key,当发送
消息到交换机后,交换机会根据 routing key 将消息发送到对应的队列。

7.消息可靠性

保证消息的可靠性需要分别从producer、broker、constomer三方面入手：

持久化
- exchange
- queue
- message
生成方确认 confirm
消费方签收 ack
Broker高可用

7.1 消息可靠性投递

在某些领域消息的可靠性必须高，特别像金融领域，涉及到金额相关的消息必须确保百分百确认，从而保证保障我们的业务系统安全、可靠、稳定的运行。RabbitMQ提供了两个功能用来控制消息的可靠性投递(confrim确认模式、 return退回模式)，通过结合以上个模式,保证了消息可以从生产者100%投递到Queue中。

消息可靠性投递解决的是从生产端到消息中间件的可靠性问题。

7.1.1 confrim确认模式

confrim确认模式保证的是producer生产者到exChange交换机过程中的消息投递。当投递成功时，会返回一个confrimCallback（true）的值；当投递失败时，则返回false。

1.在配置文件中开启消息确认机制：

//在yml文件中配置支持退回模式
spring:rabbitmq:#新版本使用这个即可 confirm模式publisher-confirm-type: correlated#老版本的话用下面这个即可开启confirm确认模式#publisher-confirms: true

2.配置RabbitMQ：

package top.xcyxiaoxiang.config;
/**  * rabbitMQ配置类  *  * @author xcy.小相  * @date 2022/11/5  22:34  */@Configuration
public class RabbitMQconfig {  @Autowired  private CachingConnectionFactory cachingConnectionFactory;  @Bean  public RabbitTemplate rabbitTemplate(){  RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);  rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {  /**  * 消息回调机制 -- confirm机制 ，producer---> broker  * @param correlationData 关联的数据，此处指的是关联的msgId  * @param ack 确认参数，消息是发布成功（producer ---> broker）  * @param cause 当发送失败时，失败原因  */  @Override  public void confirm(CorrelationData correlationData, boolean ack, String cause) {  if(ack){  System.out.println("消息成功到达Exchange");  }else{  System.out.println("消息未到达Exchange,错误原因:"+cause);  }  }        });  return  rabbitTemplate;  }  /**  * 声明交换机  * @return  */  @Bean  public Exchange bootExchange(){  return ExchangeBuilder.directExchange("boot_exchange").build();  }  /**  * 声明队列  * @return  */  @Bean  public Queue bootQueue(){  return QueueBuilder.durable("boot_queue").build();  }  /**  * 将队列绑定到交换机上。  * @return  */  @Bean  public Binding bootBinding(){  return BindingBuilder.bind(bootQueue()).to(bootExchange()).with("boot_routing_key").noargs();  }
}

3.测试生产者发送消息：

@Test
public void test01(){  String data = "hello confrim";  // 正确 交换机存在
rabbitTemplate.convertAndSend("boot_exchange","boot_routing_key",data.getBytes());]
// 异常情况 交换机不存在
rabbitTemplate.convertAndSend("boot_exchange_error","boot_routing_key",data.getBytes());  try {  // 停几秒后再关闭程序,以防出现下面的问题  // clean channel shutdown; protocol method: #method<channel.close>(reply-code=200, reply-text=OK, class-id=0, method-id=0)  Thread.sleep(3000);  } catch (InterruptedException e) {  e.printStackTrace();  }
}

7.1.2 return退回模式

return退回模式保证的是从Exchange到Queue过程中的消息投递，只有在投递失败时才会返回一个returnCallBack。
未成功路由的消息将会被返回给生产者，生产者通过returnedMessage监听回退到的消息。
1.在配置文件中开启回退模式

publisher-returns: true

2.在配置类中配置RabbiMQ的回退机制。

    @Bean
public RabbitTemplate rabbitTemplate(){……
// 设置交换机处理失败消息的模式：重新返回给生产者，生产者通过returnedMessage监听回退到的消息。
rabbitTemplate.setMandatory(true);
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {  /**  * return机制   comfirmer -----> broker，只有在未成功送达时才会执行。  * @param message 消息对象  * @param replyCode 错误码  * @param replyText 错误信息  * @param exchange 交换机名称  * @param routingKey 路由key  */    @Override  public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {  System.out.println("消息未成功到达Queue:"+replyText);  }
});
……
}

3.测试生产者发送消息

/**  * return 消息回退机制 测试  */
@Test
public void test02() {  String data = "hello confrim";  //rabbitTemplate.convertAndSend("boot_exchange","boot_routing_key",data.getBytes());  rabbitTemplate.convertAndSend("boot_exchange", "boot_routing_key_error", data.getBytes());  try {  // 停几秒后再关闭程序,以防出现下面的问题  // clean channel shutdown; protocol method: #method<channel.close>(reply-code=200, reply-text=OK, class-id=0, method-id=0)  Thread.sleep(3000);  } catch (InterruptedException e) {  e.printStackTrace();  }
}

7.2 Cutomer ACK消费确认机制

Constomer ACK机制保证的是从Queue到Constomer消费者之间的消息传递，如下图中的黄色区域。
通常情况下，处理消息的业务逻辑是不同的，为了保证消息能够在处理过程中不丢失或出现异常情况而导致处理失败，可以采用手动签收的方式进行消息确认。

当消息没有被消费且成功被消费后进行ack签收确认；当消费过程中出现异常时，拒绝签收并将消息重回队列，等待再次消费。

spring:  rabbitmq:listener:direct:  #        acknowledge-mode: auto    # confirm ack模式: 根据异常情况自动签收确认  acknowledge-mode: manual # confirm ack模式：手动签收确认  #        acknowledge-mode: none   # confirm ack模式：自动签收确认

package top.xcyxiaoxiang.constomer;
/**  * 消费者  *  * @author xcy.小相  * @date 2022/11/6  0:52  */@Component
public class Constomer {  private RabbitTemplate rabbitTemplate;  @RabbitListener(queues = "boot_queue")  public void constomerHandler(Message message, Channel channel){  MessageHeaders headers = message.getHeaders();  long tag = (long) headers.get(AmqpHeaders.DELIVERY_TAG);  // 当前消息没有被消费且成功被消费后 进行ack签收确认  try {  System.out.println("当前消息被消费:"+message.getPayload());  //int a = 10/0;  /*确认签收。  参数说明：basicAck(long deliveryTag, boolean multiple)  - deliveryTag： 当前消息的为唯一标识。  - multiple： 是否允许多条消息同时签收 一般为false。  */  channel.basicAck(tag,false);  }catch (Exception e){  System.out.println("当前消息已经被消费或消费过程中出现异常");  try {  /*  不签收消息，生产者会重新发送该消息。  参数说明：basicNack(long deliveryTag, boolean multiple, boolean requeue)  - deliveryTag: 当前消息的唯一标识  - multiple： 是否允许同时退回多条消息，一般为false。  - requeue： 是否重回队列，即让broker再次发送消息到消费者。  *                * */                channel.basicNack(tag,false,true);  } catch (IOException ex) {  ex.printStackTrace();  }  }    }
}

9. 消息可靠性保障

消息可靠性保障通过消息补偿机制来实现。

10.消息幂等性处理

幂等性操作的指任意多次执行操作所产生的影响均与一次执行的影响相同。在MQ中指的是消费多条相同的消息时，得到的结果与消费一次的结果相同。
我们通常使用乐观锁机制进行消息幂等性操作。
即消息传递过程中添加一个标识字段version,当消费者消费该消息之后，将version字段的值 +1。这是，再次消费相同的消息时，由于版本号不同就是导致消息不会被再次消费。