企业网站策划大纲模板,做网站字体一般设置,wordpress 图片墙插件,wordpress强制使用插件一、如何保证消息的可靠性 #x1f349;
1.保证消息的可靠性投递 #x1f95d;
在生产环境中由于一些不明原因#xff0c;导致 rabbitmq 重启#xff0c;在 RabbitMQ 重启期间生产者消息投递失败#xff0c;导致消息丢失#xff0c;需要手动处理和恢复。于是#xff0…一、如何保证消息的可靠性
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机制
1开启confirm #开启confirm机制 默认为none会自动删除数据 开启手动模式 correlated
spring.rabbitmq.publisher-confirm-typecorrelated
(2)设置rabbitTemplate的confirmCallback回调函数 Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void text01(){rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {Overridepublic void confirm(CorrelationData correlationData, boolean b, String s) {if (!b)System.out.println(系统繁忙请重新发送);}});rabbitTemplate.convertAndSend(zt_exchange,b.aaa, 你好阿娇);}return机制 #开启return机制 用来捕捉虚拟主机向队列中传递信息错误
spring.rabbitmq.publisher-returnstrue2设置回调 Testpublic void text02(){rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {//当交换机发送消息到队列过程中失败启动当前方法System.out.println(replyCode);}});rabbitTemplate.convertAndSend(zt_exchange,b.v, 你好阿娇);}2.如何保证消息在队列中不丢失
(1)设置队列为持久化 (2)设置消息的持久化
生产消息不设置过期时间默认为持久化 3.确保消息能可靠的被消费掉
ACK确认机制
多个消费者同时收取消息收取消息到一半突然某个消费者挂掉要保证此条消息不丢失就需要acknowledgement机制就是消费者消费完要通知服务端服务端才将数据删除
这样就解决了即使一个消费者出了问题没有同步消息给服务端还有其他的消费端去消费保证了消息不丢的case。
ACK的实现
ack指Acknowledge确认。 表示消费端收到消息后的确认方式。 有三种确认方式 自动确认acknowledge“none” 手动确认acknowledge“manual” 根据异常情况确认acknowledge“auto”这种方式使用麻烦并且不常用不作讲解
其中自动确认是指当消息一旦被Consumer接收到则自动确认收到并将相应 message 从 RabbitMQ 的消息队列中移除。但是在实际业务处理中很可能消息接收到业务处理出现异常那么该消息就会丢失。如果设置了手动确认方式则需要在业务处理成功后调用channel.basicAck()手动签收如果出现异常则调用channel.basicNack()方法让其自动重新发送消息。 #修改消费端 手动确认消息 默认自动确认 none
spring.rabbitmq.listener.simple.acknowledge-modemanual2修改代码 package com.lzq.listener;import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
Component//交于容器创建并管理
public class MyListener {RabbitListener(queues{zt_queue01})
//queues表示你监听的队列名public void hello01(Message message , Channel channel) throws IOException { //把监听到的消息封装到Mmessage类对象中long deliveryTag message.getMessageProperties().getDeliveryTag();byte[] body message.getBody();String s new String(body);System.out.println(123);// Map map JSON.parseObject(s, HashMap.class);try {System.out.println(消息的内容:s);//basicAck:确认消息 -- rabbit服务端删除/** long deliveryTag,第一个参数为消息的标志* boolean multiple 第二个参数为是否把该消费之前未确认的消息一起确认掉* */channel.basicAck(deliveryTag,true);}catch (Exception e){//basicNack:服务继续发送消息/** long deliveryTag, boolean multiple, 前两个参数与上面的意义一样* boolean requeue 是否要求rabbitmq服务器重新发送该消息* */channel.basicNack(deliveryTag,true,true);}
//业务操作}
}
总结: 如何保证消息的可靠性?
[1] 保证消息的可靠性投递: confirm机制和return机制
[2] 队列中:—持久化
[3]使用ack机制保证消费者的可靠性消费。
二、延迟队列
TTL
TTL 全称 Time To Live存活时间/过期时间。
当消息到达存活时间后还没有被消费会被自动清除。
RabbitMQ可以对消息设置过期时间也可以对整个队列Queue设置过期时间。 根据设置的时间十秒后会消失 Testpublic void test04(){/*** String exchange, String routingKey, Object message,* CorrelationData correlationData*/Message message new Message(咻咻咻.getBytes());message.getMessageProperties().setExpiration(10000);rabbitTemplate.send(bbb,a.b,message);}小结:
设置队列过期时间使用参数x-message-ttl单位ms(毫秒)会对整个队列消息统一过期。
设置消息过期时间使用参数expiration。单位ms(毫秒)当该消息在队列头部时消费时会单独判断这一消息是否过期。
如果两者都进行了设置以时间短的为准。
死信队列
死信队列英文缩写DLX 。Dead Letter Exchange死信交换机当消息成为Dead message后可以被重新发送到另一个交换机这个交换机就是DLX。 什么样的消息会成为死信消息 队列消息长度到达限制 消费者拒接消费消息basicNack/basicReject,并且不把消息重新放入原目标队列,requeuefalse 原队列存在消息过期设置消息到达超时时间未被消费
队列绑定死信交换机 延迟队列
延迟队列即消息进入队列后不会立即被消费只有到达指定时间后才会被消费。
需求 下单后30分钟未支付取消订单回滚库存。 新用户注册成功7天后发送短信问候。
实现方式 定时器性能差—每隔一段时间要进行数据库查询。 延迟队列
通过消息队列完成延迟队列的功能
很可惜在RabbitMQ中并未提供延迟队列功能。
但是可以使用TTL死信队列 组合实现延迟队列的效果。 三、 如何防止消费者重复消费消息
消息的幂等性—无论操作几次结果都是一样。
1、生成全局id存入redis或者数据库在消费者消费消息之前查询一下该消息是否有消费过。
2、如果该消息已经消费过则告诉mq消息已经消费将该消息丢弃手动ack。
3、如果没有消费过将该消息进行消费并将消费记录写进redis或者数据库中。 简单描述一下需求如果订单完成之后需要为用户累加积分又需要保证积分不会重复累加。那么再mq消费消息之前先去数据库查询该消息是否已经消费如果已经消费那么直接丢弃消息。
生产者
package com.ykq.score.producer;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;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: 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;}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);}}
}消费者
package com.xiaojie.score.consumer;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;Component
Slf4j
public class ScoreConsumer {Autowiredprivate ScoreMapper scoreMapper;RabbitListener(queues {ykq_score_queue})public void onMessage(Message message, Headers MapString, 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(消费失败采取相应的人工补偿);} }
}
