如何保证消息的可靠性

在生产环境中由于一些不明原因，导致 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机制

（1）开启confirm

ini复制代码#开启confirm机制  默认为none
spring.rabbitmq.publisher-confirm-type=correlated

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

typescript复制代码@Autowired
private RabbitTemplate rabbitTemplate;

@Test
void test02(){
    rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            //生产者向交换机发送信息的时候会触发该方法,ack如果是true代表发送成功，false代表发送失败
            if(!ack){
            //这里实际开发中，根据项目需求，完成相应的操作
                rabbitTemplate.convertAndSend("topics_exchange","three.one","这个交换机是真实存在的");
            }
        }
    });

    rabbitTemplate.convertAndSend("topics_exchang","three.one","我故意发送给了一个不存在的交换机");

}

return机制

(1)开启retrun

ini复制代码#开启return机制
spring.rabbitmq.publisher-returns=true

(2)设置交换机到队列之间的回调函数

typescript复制代码@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void test03(){
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             *
             * @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) {
                //当交换机到队列分发消息时会触发，这里通常是根据routingKey进行传输
                System.out.println("message:"+message);
                System.out.println("replyCode:"+replyCode);
                System.out.println("replyText:"+replyText);
                System.out.println("exchange:"+exchange);
                System.out.println("routingKey:"+routingKey);

            }
        });
        
    rabbitTemplate.convertAndSend("topics_exchange","aaa.bbb","我故意发送给了一个不存在的routingKey");
}

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

(1)设置队列为持久化

(2)设置消息的持久化

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

ACK确认机制

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

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

ACK的实现

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

有三种确认方式：自动确认：acknowledge="none"

手动确认：acknowledge="manual"

根据异常情况确认：acknowledge="auto"，（这种方式使用麻烦，并且不常用，不作讲解）

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息队列中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

修改消费端为手动确认消息

ini复制代码#修改消费端为手动确认信息，默认的是自动确认：none
spring.rabbitmq.listener.simple.acknowledge-mode=manual

设置消费端的确认机制在业务代码执行之后

java复制代码
//交予spring容器管理
@Component
public class MyListener {

    //basicAck：确认消息--rabbit服务端删除
    //basicNack：如果出现异常情况，服务继续发送消息
    @RabbitListener(queues = {"topics-queues002"})
    public void topics(Message message, Channel channel) throws IOException {
        //消息的标志
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        final byte[] body = message.getBody();
        String s = new String(body);
        System.out.println("消息的内容是"+s);
        try{
            System.out.println("业务代码逻辑执行完毕");
            /**
             *long deliveryTag, 消息的标签
             * boolean multiple  是否把该消息之前未确定的消息一起确认掉
             */
            //确认消息方法
            channel.basicAck(deliveryTag,true);
        }catch (Exception e){
            /**
             * long deliveryTag,boolean multiple,
             * boolean requeue  是否要求rabbitmq服务重新发送该消息
             */
            //服务继续发送消息方法
            channel.basicNack(deliveryTag,false,true);

        }


    }
}

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

[1] 保证消息的可靠性投递: confirm机制和return机制

[2] 队列中:---持久化

[3]使用ack机制保证消费者的可靠性消费。

延迟队列

TTL

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

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

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

java复制代码@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void test04(){
    /**
     * 我在创建队列中设置了TTL【过期时间】为20s，而这里设置了过期时间为40s---经过测试发现，队列会按照时间短的执行。
     */
    final Message message = new Message("我要40秒过期".getBytes());
    message.getMessageProperties().setExpiration("40000");

    rabbitTemplate.convertAndSend("timeout-exchange","",message);
}

小结:

设置队列过期时间使用参数：x-message-ttl，单位：ms(毫秒)，会对整个队列消息统一过期。

设置消息过期时间使用参数：expiration。单位：ms(毫秒)，当该消息在队列头部时（消费时），会单独判断这一消息是否过期。

如果两者都进行了设置，以时间短的为准。

死信队列

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

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

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

队列绑定死信交换机

延迟队列

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

需求：

1.下单后，30分钟未支付，取消订单，回滚库存。

2.新用户注册成功7天后，发送短信问候。

实现方式：

1.定时器：性能差---每隔一段时间要进行数据库查询。

2.延迟队列

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

很可惜，在RabbitMQ中并未提供延迟队列功能。

但是可以使用：TTL+死信队列组合实现延迟队列的效果。

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

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

1、生成全局id，存入redis或者数据库，在消费者消费消息之前，查询一下该消息是否有消费过。

2、如果该消息已经消费过，则告诉mq消息已经消费，将该消息丢弃（手动ack）。

3、如果没有消费过，将该消息进行消费并将消费记录写进redis或者数据库中。

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

生产者

typescript复制代码@Component
@Slf4j
public class ScoreProducer implements RabbitTemplate.ConfirmCallback {
    @Autowired
    private 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: 2023/7/10 22:30
     */
    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: 2023/7/10 22:22
     */
 
    @Async
    public 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;
        });
    }
 
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String s) {
        if (ack) {
            log.info(">>>>>>>>消息发送成功:correlationData:{},ack:{},s:{}", correlationData, ack, s);
        } else {
            log.info(">>>>>>>消息发送失败{}", ack);
        }
    }
}

消费者

less复制代码@Component
@Slf4j
public class ScoreConsumer {
    @Autowired
    private 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("消费失败，采取相应的人工补偿");
        } 
    }
}

原文链接：https://juejin.cn/post/7255197799049347127

rabbitMQ的高级--如何保证消息的可靠性--延迟队列--消息的幂等性

如何保证消息的可靠性

confirm机制

（1）开启confirm

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

return机制

(1)开启retrun

(2)设置交换机到队列之间的回调函数

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

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

修改消费端为手动确认消息

设置消费端的确认机制在业务代码执行之后

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

延迟队列

TTL

小结:

死信队列

队列绑定死信交换机

延迟队列

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

继续阅读

昨天，对同步记录的细节又完善了下。把消息中间件中，关于图片的处理和回写，挑了最后的处理结果记录了下来。这样就可以完整地观

SpringBoot实现Rabbit消息队列

RabbitMQ 高级特性 - 延迟队列

springboot集成rabbitmq 延迟队列

RabbitMQ--过期时间TTL、死信队列、延迟队列和优先队列

SpringBoot+RabbitMq 采用 TTL（延迟队列）+ DLX（死信队列）实现延迟消费1、pom文件引入jar包2、yml 配置3、声明交换机、队列及绑定4、定义常量类5、发送消息6、消费消息

springboot+rabbitmq+websocket集群推送解决方案

兔、am。迷迷糊糊中读书人。cat：cat。black。g。purpurple。light。white。Sir。brow

【RocketMQ】延迟消息(延迟队列)1. 什么是延迟消息2. 示例3. 原理参考

go语言操作rabbitmq

Redis实战篇：巧用Zset实现延迟队列功能

#微头条创作计划#al作画#最美的画图##个人小画分享#中文翻译成英文后让al读英文作画Setaria,wonderfu

RabbitMQ延时队列之一

11rabbit

Springboot 指定重发的次数和延迟时间，定时异步执行重发任务