备案的网站程序上传,wordpress页眉设置,国际论坛网站模板,展览展厅设计制作消息丢失的可能性 支付服务先扣减余额和更新支付状态#xff08;这俩是同步调用#xff09;#xff0c;然后通过RabbitMq异步调用支付服务更新订单状态。但是有些情况下#xff0c;可能订单已经支付 #xff0c;但是更新订单状态却失败了#xff0c;这就出现了消息丢失。…消息丢失的可能性 支付服务先扣减余额和更新支付状态这俩是同步调用然后通过RabbitMq异步调用支付服务更新订单状态。但是有些情况下可能订单已经支付 但是更新订单状态却失败了这就出现了消息丢失。
发送者在发送的过程中出现了网络故障RabbitMQ在发送消息的过程中出现了问题消费者在更新订单状态的时候出现了问题
发送者的可靠性
发送者确认机制需要与MQ进行通信和确认会影响消息发送的效率且一般出现的概率极低所以一般不用这个。 方法1. 发送者重连
确保发送者与MQ之间连接的可靠性。有的时候由于网络波动可能出现发送者连接MQ失败的情况这个配置是关闭的可以开启连接失败后的重连机制
spring:rabbitmq:connection-timeout: 1s # 设置MQ的连接超时时间template:retry:enabled: true # 开启超时重试机制(默认是false)initial-interval: 1000ms # 失败后的初始等待时间multiplier: 1 # 失败后下次的等待时长倍数下次等待时长 initial-interval * multipliermax-attempts: 3 # 最大重试次数【注】当网络不稳定时利用重试机制可以提高消息发送的成功率但是SpringAMQP提供的重试机制是阻塞式的重试如果需要多次重试等待当前线程被阻塞会影响性能。 如果对业务性能有要求建议禁用重试机制如果一定要使用要合理的配置等待时常和重试次数或使用异步线程来执行发送消息的代码。 方法2. 发送者确认
确保消息发送的可靠性。SpringAMQP提供了Publisher Confirm和Publisher Return两种机制开启确认机制后当发送者发送消息给MQ后MQ会返回确认结果给发送者返回的结果有以下几种情况
消息投递到MQ但是路由失败此时通过PublisherReturn返回路由异常信息然后返回ACK告知投递成功。例如 消息发送给图中的exchange1但是RoutingKey写错了没有匹配到正确的队列也会导致路由失败。消息发送给图中的exchange2但是它底下没有绑定新的队列就会导致路由失败。 临时消息【不需要往磁盘做持久化的消息】投递到MQ并入队成功返回ACK告知投递成功。持久消息投递到MQ并入队完成持久化返回ACK告知投递成功。其他情况都会返回NACK告知投递失败。 步骤
在发送方publisher所在的微服务的application.yml中配置
spring:rabbitmq:publisher-confirm-type: correlated # 开启publisher confirm机制并设置confirm类型publisher-returns: true # 开启publisher return机制publisher-confirm-type有三种模式 none关闭confirm机制simple同步阻塞等待MQ回执消息correlatedMQ异步回调方式返回回执消息常用 开启回调机制每个RabbitTemplate只能配置一个ReturnCallback在发送者publisher所在的项目启动时配置即可。
Configuration
RequiredArgsConstructor
Slf4j
public class MqConfig {private final RabbitTemplate rabbitTemplate;PostConstruct // 在Bean初始化完成后调用这个方法只会调用一次public void init() {// 返回ACK但是此时路由失败就会走这个方法rabbitTemplate.setReturnsCallback(returnedMessage - {log.error(监听到了消息return callback);log.debug(exchange: {}, returnedMessage.getExchange());log.debug(routingKey: {}, returnedMessage.getRoutingKey());log.debug(message: {}, returnedMessage.getMessage());log.debug(replyCode: {}, returnedMessage.getReplyCode());log.debug(replyText: {}, returnedMessage.getReplyText());});}
}开启消息确认机制发送消息、指定消息ID、每次发送消息都需要配置一个ConfirmCallback
public void testConfirmCallback() {CorrelationData cd new CorrelationData(UUID.randomUUID().toString());cd.getFuture().addCallback(new ListenableFutureCallbackCorrelationData.Confirm() {// 【几乎不可能发生】Future发生异常时的处理逻辑Overridepublic void onFailure(Throwable ex) {log.error(spring ampq处理确认结果异常, ex);}// 成功拿到MQ结果判断是ACK还是NACKOverridepublic void onSuccess(CorrelationData.Confirm result) {if(result.isAck()) {// ACKlog.debug(收到ACK消息发送成功);}else {// NACKlog.debug(收到NACK消息发送失败失败原因{},result.getReason());}}});rabbitTemplate.convertAndSend(hmall.direct, blue, hello world, cd); // 发送消息
}MQ的可靠性
RabbitMQ一般会将收到的信息保存到内存速度快中降低消息收发的延迟这样会导致
MQ宕机内存中的消息会丢失。内存空间有限消费者故障或处理过慢会导致消息积压引发MQ阻塞。 【案例】发送者往MQ发消息MQ会把数据保存到内存中如果内存满了MQ就会把一部分数据迁移到磁盘中暂时进行持久化存储移动到磁盘的这段时间发送者发送的消息就会产生丢失。 方法1. 数据持久化
数据持久化就是把数据持久化到磁盘但是不是向上边那个案例等满了再去持久化被动而是提前进行持久化。
交换机的持久化默认开启的 队列的持久化默认开启的 消息持久化默认是非持久的 在发送消息的时候设定的 【案例】比较一下持久化和非持久化的性能。 发100w条消息给MQ 这是非持久化的方式使用纯内存的方式存储每次内存满之后MQ就会把消息写到磁盘中此时就会出现阻塞状态处理速度降低到0 【问题】可能出现消息丢失和MQ阻塞 【解决办法】使用持久化的方式
public void testSendPersistentMsg() {// 自定义构建消息Message msg MessageBuilder.withBody(hello world.getBytes(StandardCharsets.UTF_8)) // 消息体.setDeliveryMode(MessageDeliveryMode.PERSISTENT) // 投递模式持久化.build();for (int i 0; i 100000; i) {rabbitTemplate.convertAndSend(simple.queue, msg);}
}Mq并没有阻塞每发一条消息就赶紧把它存到磁盘中和纯内存方式相比不会有个中断的过程。
方法2. Lazy Queue推荐
【问题】由于使用了消息持久化的方式发到MQ的消息不仅要到内存还要在磁盘中写一份这会导致整体的并发能力下降 【特征】
接收到消息后直接入磁盘不再存储到内存在写磁盘的时候也对写入磁盘的操作进行一些优化比传统的写操作高很多消费者要消费消息时才会从磁盘中读取并加载到内存 【问题】可能会影响消费者处理消息的速度【解决】可以提前缓存部分消息到内存最多2048条
控制台声明Lazy Queue队列 Java代码添加
声明Bean
Bean
public Queue lazyQueue(){return QueueBuilder.durable(lazy.queue).lazy() // 开启Lazy模式.build();
}RabbitListener注解
RabbitListener(queuesToDeclare Queue(name lazy.queue,durable true,arguments Argument(name x-queue-mode, value lazy) // 开启Lazy模式
))
public void listenLazyQueue(String msg){log.info(接收到 lazy.queue的消息{}, msg);
}消费者的可靠性
消费者确认机制
为了确认消费者是否成功处理消息当消费者处理消息结束后应该向MQ发送一个回执告知MQ自己的消息处理状态。有如下几种消息处理状态
ack处理消息成功RabbitMQ从队列中删除该消息nack消息处理失败RabbitMQ需要再次投递消息reject消息处理失败并拒绝该消息RabbitMQ从队列中删除该消息【在处理的过程中发现消息的内容有问题没有重试的必要直接拒绝就行】 【注意】不管是哪种情况都应该等消息处理完后得到结果再返回不要一拿到消息就返回 返回消息处理状态的过程类似于处理事务事务处理成功返回ACK处理失败返回NACK SpringAMQP允许通过在消费者的配置文件选择ACK的处理方式有三种
none不处理消息投递给消费者后立刻ack消息会立刻从MQ中删除别用manual手动模式需要在业务代码中调用api发送ack或reject存在业务入侵但是更灵活。auto自动模式利用AOP对消息处理逻辑进行了环绕增强 业务处理正常自动返回ack业务处理异常自动返回nack消息处理或校验异常【MessageConversionException】自动返回reject
spring:rabbitmq:listener:simple:acknowledge-mode: auto# 不做处理失败重试策略
在消费者出现异常时利用本地重试而不是无限的重新入队到mq可以在消费者的yaml文件中添加配置来开启重试机制。
spring:rabbitmq:listener:simple:retry:enabled: true # 开启消费者失败重试initial-interval: 1000ms # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数下次等待时长 multiplier * last-intervalmax-attempts: 3 # 最大重试次数达到最大重试次数后MQ会把消息丢弃stateless: true # true无状态false有状态。如果业务中包含事务这里改为false【问题】在开启重试模式后重试次数耗尽如果消息仍然失败默认会把消息进行丢弃。 【解决】因此需要有MessageRecoverer接口来处理包含三种不同的实现
RejectAndDontRequeueRecoverer默认重试耗尽后直接reject丢弃消息。ImmediateRequeueMessageRecoverer重试耗尽后返回nack消息重新入队。RepublishMessageRecoverer重试耗尽后将失败消息投递到指定的交换机 。
修改失败重试策略为RepublishMessageRecoverer
定义接收失败的交换机、队列、定义RepublishMessageRecoverer
Configuration
public class ErrorMessageConfiguration {// 定义接收失败的交换机Beanpublic DirectExchange errorExchange() {return new DirectExchange(error.direct);}// 定义接收失败的队列Beanpublic Queue errorQueue() {return new Queue(error.queue);}// 定义绑定关系Beanpublic Binding errorQueueBinding() {return BindingBuilder.bind(errorQueue()).to(errorExchange()).with(error);}// 定义失败处理策略Beanpublic MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {return new RepublishMessageRecoverer(rabbitTemplate, error.direct, error);}
}业务幂等性
f(x) f(f(x))指同一个业务执行一次或多次对业务状态的影响是一致的。
幂等业务查询业务、删除业务非幂等业务用户下单需要扣减库存、用户退款业务需要恢复余额
方案1. 唯一消息id
给每个消息设置一个唯一id利用id区分是否是重复消息
每条消息都生成一个唯一id与消息一起投递给消费者消费者接收到消息后处理自己的业务业务处理成功后将消息id保存到数据库中如果下次又收到相同消息去数据库查询判断是否存在存在则视为重复消息放弃处理
在发送方配置Bean用来自动创建消息id
Configuration
public class MqConfig {Beanpublic MessageConverter messageConverter() {Jackson2JsonMessageConverter converter new Jackson2JsonMessageConverter();converter.setCreateMessageIds(true); // 配置自动创建消息idreturn converter;}
}在接收方接收消息id
RabbitListener(queues simple.queue)public void listenSimpleQueue(Message msg) { // 使用字符串发送就用字符串接收log.info(监听到simple.queue的消息{}, msg);log.info(消息id{}, msg.getMessageProperties().getMessageId());// throw new RuntimeException(故意的);}方案2. 业务判断常用
结合业务逻辑基于业务本身做判断。 【案例】当用户下单成功后通过MQ通知交易服务来修改订单状态为已支付这里记作消息1修改成功后交易服务返回ACK给MQ此时出现了网络的故障MQ没有收到交易服务发送的ACKMQ认为交易服务宕机消息又重新入队。 就在此刻用户点击了申请退款直接向交易服务修改订单状态为退款中这个操作没有走MQ此时订单状态是退款中但是消息1还在消息队列中。 此时网络恢复了MQ又将消息1发送给交易服务此时交易服务又把订单状态标记为已支付订单申请退款中的状态又被覆盖了。 【解决】通知来的时候先判断订单的状态再进行操作。
Component
RequiredArgsConstructor
public class PayStatusListener {private final IOrderService orderService;RabbitListener(bindings QueueBinding(value Queue(name trade.pay.success.queue, durable true),exchange Exchange(name pay.direct),key pay.success))public void listenPaySuccess(Long orderId) {// 1.查询订单Order order orderService.getById(orderId);// 2.判断订单状态是否为未支付if(order null || order.getStatus() ! 1) {// 不做处理return;}// 3.标记订单状态为已支付orderService.markOrderPaySuccess(orderId);}
}延迟消息
延迟消息发送者发送消息时指定一个时间消费者不会立刻收到消息而是在指定时间之后才收到消息。 延迟任务设置在一定时间后才执行的任务。
方案1. 死信交换机
当一个队列中的消息满足下列情况之一的就会成为死信
消费者使用basic.reject或basic.nack声明消费失败并且消息的requeue参数设置为false。消息是一个过期消息达到队列设置的过期时间 或 消息本身设置的过期时间超时无人消费。要投递的队列消息堆积满了最早的消息可能成为死信。
队列通过dead-letter-exchange属性指定了一个交换机那么该队列中的死信就会投递到这个交换机中这个交换机就叫做死信交换机DLX。
声明死信队列、死信交换机、它们之间的绑定关系
RabbitListener(bindings QueueBinding(value Queue(name dlx.queue, durable true), // 死信队列exchange Exchange(name dlx.direct, type ExchangeTypes.DIRECT), // 死信交换机key {hi} ))public void listenDlxQueue(String msg) {log.info(消费者监听到dlx.queue的消息: msg);}声明普通队列、普通交换机、它们之间的绑定关系并把队列绑定到死信交换机上此时就不需要把它绑定消费者了
Configuration
public class NormalConfiguration {Beanpublic DirectExchange normalExchange() { // 普通交换机return ExchangeBuilder.directExchange(normal.direct).build();}Beanpublic Queue normalQueue() { // 普通队列return QueueBuilder.durable(normal.direct) // 队列名字.deadLetterExchange(dlx.direct) // 死信交换机名字.build();}Beanpublic Binding normalQueueBinding(Queue normalQueue, DirectExchange normalExchange) { // 绑定关系// 把队列绑定到交换机return BindingBuilder.bind(normalQueue) // 队列.to(normalExchange) // 交换机.with(hi);// 这里绑定关系要和普通队列的绑定关系保持一致}
}
发送延迟消息
Test
public void testSendDelayMsg() {rabbitTemplate.convertAndSend(normal.direct, hi, hello world, message - {// 当消息被转成Message对象后还可以进一步做加工message.getMessageProperties().setExpiration(10000); // 设置消息过期时间(10s)return message;});
}【注】normal.direct和normal.queue之间绑定的BindingKey 与 dlx.direct和dlx.queue之间绑定的BindingKey要一致 方案2. 延迟消息插件DelayExchange推荐
这个插件可以将普通交换机改造为支持延迟消息功能的交换机当消息投递到交换机后可以暂存一段时间到后期再投递到队列。
一、安装插件
插件下载地址DelayExchange需要把插件放在RabbitMQ插件目录对应的数据卷下
docker volume inspect mq-plugins执行命令安装插件
docker exec -it rabbitmq rabbitmq-plugins enable rabbitmq_delayed_message_exchange二、使用插件
声明延迟交换机只要设置delay的属性为true即可
RabbitListener(bindings QueueBinding(value Queue(name delay.queue, durable true),exchange Exchange(name delay.direct, delayed true, type ExchangeTypes.DIRECT), // 只要设置一个delayed属性为true即可key {hi}
))
public void listenDelayQueue(String msg) {log.info(消费者监听到delay.queue的消息: msg);
}发送延迟消息通过消息头x-delay来设置过期时间
Test
public void testSendDelayMsgByPlugin() {rabbitTemplate.convertAndSend(delay.direct, delay, hello world, message - {message.getMessageProperties().setDelay(10000);// 添加延迟消息属性return message;});
}延迟消息的实现需要记录消息的过期时间计时的时钟需要依赖cpu是个cpu密集型任务。因此使用延迟消息时需要避免同一时刻在mq里存在大量的延迟消息尽可能地让延迟消息的延迟时间不要太长。
