1.如何保证消息的可靠性

1.1.消息的可靠投递

confirm机制

return机制

1.2.如何保证消息在队列中不丢失

1.3.确保消息能可靠的被消费掉

2.延迟队列

2.1.TTL

2.2.死信队列

2.3.延迟队列

3.如何防止消费者重复消费消息

1.如何保证消息的可靠性

1.1.消息的可靠投递

在生产环境中由于一些不明原因，导致 rabbitmq 重启，在 RabbitMQ 重启期间生产者消息投递失败，导致消息丢失，需要手动处理和恢复。于是，我们开始思考，如何才能进行 RabbitMQ 的消息可靠投递呢？特别是在这样比较极端的情况，RabbitMQ 集群不可用的时候，无法投递的消息该如何处理呢？

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。

confirm 确认模式
return 退回模式
消息从 producer 到 exchange 则会返回一个 confirmCallback 。
消息从 exchange-->queue 投递失败则会返回一个 returnCallback 。

默认rabbitmq不开启上面两种模式

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

confirm和return的实现

设置ConnectionFactory的publisher-confirm-type: correlated开启确认模式。

使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。

设置ConnectionFactory的publisher-returns="true" 开启退回模式。

使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后执行回调函数returnedMessage。

confirm机制

演示:4.springboot整合RabbitMQ

(1).配置文件中开启confirm

#开启confirm确认机制
spring.rabbitmq.publisher-confirm-type=correlated

(2)设置rabbitTemplate的confirmCallback回调函数

    /*** 使用confirm机制：* （1）需要开启confirm机制。-----配置文件中加入：spring.rabbitmq.publisher-confirm-type=correlated* (2)为rabbitTemplate指定setConfirmCallback回调函数*/@Testvoid test001() {//只能保证消息从生产者到交换机的可靠性rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {//触发该方法if (ack == false) {System.out.println("未来根据项目的需要，完成相应的操作");}}});rabbitTemplate.convertAndSend("Topics-exchange", "lazy.aaa", "Hello RabbitMQ...");}

return机制

(1).配置文件中开启return

#开启return机制
spring.rabbitmq.publisher-returns=true

(2)设置rabbitTemplate的return回调函数

    /*** return机制：* （1）开启return机制---配置文件加入：spring.rabbitmq.publisher-returns=true* （2）为rabbitTemplate设置return的回调函数*/@Testvoid test002() {//只有当消息无法从交换机到队列时才会触发rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {//为交换机到队列分发消息失败时会触发System.out.println("replyCode====" + replyCode);System.out.println("replyText====" + replyText);}});rabbitTemplate.convertAndSend("Topics-exchange", "lazy.aaa", "Hello RabbitMQ...");}

1.2.如何保证消息在队列中不丢失

(1)设置队列为持久化

(2)设置消息的持久化

1.3.确保消息能可靠的被消费掉

ACK确认机制

多个消费者同时收取消息，收取消息到一半，突然某个消费者挂掉，要保证此条消息不丢失，就需要acknowledgement机制，就是消费者消费完要通知服务端，服务端才将数据删除

这样就解决了，即使一个消费者出了问题，没有同步消息给服务端，还有其他的消费端去消费，保证了消息不丢的case。

ACK的实现

ack指Acknowledge，确认。表示消费端收到消息后的确认方式。

有三种确认方式：

自动确认：acknowledge="none"
手动确认：acknowledge="manual"
根据异常情况确认：acknowledge="auto"（这种方式使用麻烦，并且不常用）

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息队列中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。

如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

消费端:

(1).修改消费端----手动确认消息
#修改消费端----手动确认消息
spring.rabbitmq.listener.simple.acknowledge-mode=manual
(2).修改代码
package com.wqg.listener;import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.amqp.core.Message;import java.io.IOException;/*** @ fileName:MyListener* @ description:* @ author:wqg* @ createTime:2023/7/12 18:57*/
@Component
public class MyListener {//basicAck:确认消息----rabbit服务端删除//basicNack：服务继续发送消息@RabbitListener(queues = {"Topics-queue002"})//queues:表示你监听的队列名public void h(Message message , Channel channel) throws IOException {long deliveryTag = message.getMessageProperties().getDeliveryTag();//把监听到的消息封装到Message类对象中byte[] body = message.getBody();String s = new String(body);System.out.println("消息内容==="+s);try {//int a = 10/0; //模拟宕机System.out.println("核心业务的处理~~~");/***long deliveryTag： 消息的标注* boolean multiple：是否把该消息之前未确认的消息一起确认掉*/channel.basicAck(deliveryTag,true);//确认消息} catch (Exception e) {/*** long deliveryTag; boolean multiple;* boolean requeue：是否要求rabbitmq服务重新发送该消息*/channel.basicNack(deliveryTag,true,true);}}
}

总结: 如何保证消息的可靠性?

保证消息的可靠性投递: confirm机制和return机制
队列中:---持久化
使用ack机制保证消费者的可靠性消费。

2.延迟队列

2.1.TTL

TTL 全称 Time To Live（存活时间/过期时间）。

当消息到达存活时间后，还没有被消费，会被自动清除。

RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。

演示：

使用图形化界面创建

生产者测试

结果

队列设置了过期时间而消息也设置了过期时间-----按照时间短的执行

小结:

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

2.2.死信队列

死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机），当消息成为Dead message后，可以被重新发送到另一个交换机，这个交换机就是DLX。

什么样的消息会成为死信消息：

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

队列绑定死信交换机：

演示：

使用图形化界面创建：

生产者测试

结果

2.3.延迟队列

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

需求：

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

实现方式：

定时器：性能差---每隔一段时间要进行数据库查询。
延迟队列

通过消息队列完成延迟队列的功能：

在RabbitMQ中并未提供延迟队列功能。
但是可以使用：TTL+死信队列组合实现延迟队列的效果。

演示：

队列为空

创建springboot项目

配置文件


#rabbitmq的配置
spring.rabbitmq.host=192.168.75.129
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

(1). 引入依赖

<dependencies><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.83</version></dependency><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><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency></dependencies>

(2).创建OrderController.java模拟订单

package com.wqg.controller;import com.alibaba.fastjson.JSON;
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.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.HashMap;
import java.util.UUID;/*** @ fileName:OrderController* @ description:订单* @ author:wqg* @ createTime:2023/7/13 16:24*/
@RestController
@RequestMapping("/order")
public class OrderController {@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/saveOrder")public String save(Integer pid, Integer num) {//生成一个订单号String orderId = UUID.randomUUID().toString().replace("-", "");System.out.println("下单成功，订单号为===" + orderId);HashMap<Object, Object> map = new HashMap<>();map.put("orderId", orderId);map.put("pid", pid);map.put("num", num);rabbitTemplate.convertAndSend("pt_exchange", "qy165.aaa", JSON.toJSONString(map));return "下单成功";}
}

(3).创建MyListener.java模拟监听

package com.wqg.listener;import com.alibaba.fastjson.JSON;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.HashMap;/*** @ fileName:MyListener* @ description:* @ author:wqg* @ createTime:2023/7/13 16:35*/
@Component
public class MyListener {@RabbitListener(queues = {"dead_queue"})public void hello(Message message){byte[] body = message.getBody();String s = new String(body);HashMap hashMap = JSON.parseObject(s, HashMap.class);System.out.println("message==="+hashMap);//取消订单System.out.println("取消订单号为==="+hashMap.get("orderId"));}
}

测试

控制台

3.如何防止消费者重复消费消息

消息的幂等性---无论操作几次结果都是一样。

生成全局id，存入redis或者数据库，在消费者消费消息之前，查询一下该消息是否有消费过。
如果该消息已经消费过，则告诉mq消息已经消费，将该消息丢弃（手动ack）。
如果没有消费过，将该消息进行消费并将消费记录写进redis或者数据库中。

简单描述一下需求，如果订单完成之后，需要为用户累加积分，又需要保证积分不会重复累加。那么再mq消费消息之前，先去数据库查询该消息是否已经消费，如果已经消费那么直接丢弃消息。

演示：

生产者

import com.alibaba.fastjson.JSONObject;
import com.xiaojie.score.entity.Score;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;import java.util.UUID;/*** @author * @version 1.0* @description:发送积分消息的生产者* @date*/
@Component
@Slf4j
public class ScoreProducer implements RabbitTemplate.ConfirmCallback {@Autowiredprivate RabbitTemplate rabbitTemplate;//定义交换机private static final String SCORE_EXCHANGE = "ykq_score_exchaneg";//定义路由键private static final String SCORE_ROUTINNGKEY = "score.add";/*** @description: 订单完成* @param:* @return: java.lang.String* @author xiaojie* @date: */public String completeOrder() {String orderId = UUID.randomUUID().toString();System.out.println("订单已完成");//发送积分通知Score score = new Score();score.setScore(100);score.setOrderId(orderId);String jsonMSg = JSONObject.toJSONString(score);sendScoreMsg(jsonMSg, orderId);return orderId;}/*** @description: 发送积分消息* @param:* @param: message* @param: orderId* @return: void* @author * @date:*/@Asyncpublic void sendScoreMsg(String jsonMSg, String orderId) {this.rabbitTemplate.setConfirmCallback(this);rabbitTemplate.convertAndSend(SCORE_EXCHANGE, SCORE_ROUTINNGKEY, jsonMSg, message -> {//设置消息的id为唯一message.getMessageProperties().setMessageId(orderId);return message;});}@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String s) {if (ack) {log.info(">>>>>>>>消息发送成功:correlationData:{},ack:{},s:{}", correlationData, ack, s);} else {log.info(">>>>>>>消息发送失败{}", ack);}}
}

消费者

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.xiaojie.score.entity.Score;
import com.xiaojie.score.mapper.ScoreMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.util.Map;/*** @author * @version 1.0* @description: 积分的消费者* @date */
@Component
@Slf4j
public class ScoreConsumer {@Autowiredprivate ScoreMapper scoreMapper;@RabbitListener(queues = {"ykq_score_queue"})public void onMessage(Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException {String orderId = message.getMessageProperties().getMessageId();if (StringUtils.isBlank(orderId)) {return;}log.info(">>>>>>>>消息id是：{}", orderId);String msg = new String(message.getBody());Score score = JSONObject.parseObject(msg, Score.class);if (score == null) {return;}//执行前去数据库查询，是否存在该数据，存在说明已经消费成功，不存在就去添加数据，添加成功丢弃消息Score dbScore = scoreMapper.selectByOrderId(orderId);if (dbScore != null) {//证明已经消费消息，告诉mq已经消费,丢弃消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);return;}Integer result = scoreMapper.save(score);if (result > 0) {//积分已经累加，删除消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);return;} else {log.info("消费失败，采取相应的人工补偿");} }
}