织梦dedecms网站内容页,网站模版上线需要什么意思,中国移动网站备案管理系统,搭建wordpress需要php环境吗为什么要保证消息的可靠性#xff1f; 当MQ向消费者发送一个消息之后需要得到消费者的状态#xff0c;因为消息并不一定就真的被消费者给消费了#xff0c;可能在消费的过程中出现了一些意外#xff0c;比如 1. 网络问题 2. 消息转换有问题 3. 消费者本身的业务处理有问题 … 为什么要保证消息的可靠性 当MQ向消费者发送一个消息之后需要得到消费者的状态因为消息并不一定就真的被消费者给消费了可能在消费的过程中出现了一些意外比如 1. 网络问题 2. 消息转换有问题 3. 消费者本身的业务处理有问题 消费者确认机制 消费者消息处理状态 ack:消息成功接收并且成功被处理MQ将此消息删除nack消息处理失败需要MQ重新发送消息reject消息处理失败并且拒绝该消息MQ将此消息删除 由于消息回执的处理代码比较统一因此SpringAMQP帮我们实现了消息确认。并允许我们通过配置文件设置ACK处理方式有三种模式 none不处理。即消息投递给消费者后立刻ack消息会立刻从MQ删除。非常不安全不建议使用 manual手动模式。需要自己在业务代码中调用api发送ack或reject存在业务入侵但更灵活 auto自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强当业务正常执行时则自动返回ack. 当业务出现异常时根据异常判断返回不同结果 如果是业务异常会自动返回nack 如果是消息处理或校验异常自动返回reject; 在配置文件中通过下面的配置即可设置ACK的处理方式 spring:rabbitmq:listener:simple:acknowledge-mode: c # none 默认 auto 自动确认 manual 手动确认 消费者重试机制 消费者接收了一个消息但是在处理的过程中出现异常了那么AMQP会不断的重试直到把资源占完然后崩掉这个时候就必须要设置重试机制限制重试的次数避免无限制重试。 spring:rabbitmq:listener:simple:retry:enabled: true # 开启消费者失败重试initial-interval: 1000ms # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数下次等待时长 multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态false有状态。如果业务中包含事务这里改为false 测试 先别配置重试机制然后在需要在接收消息的地方手动抛出一个异常查看控制台就会看见消费者在尝试不断的获取消息但是一直获取不到无限制的重试 RabbitListener(bindings QueueBinding(value Queue(name fanout.hamll.query1, // 队列名称durable true), // 是否持久exchange Exchange(name fanout.hamll, type ExchangeTypes.FANOUT) // 交换机名称))public void query2(String message) {System.err.println(fanout.hamll.query1 消息内容 message); throw new RuntimeException(故意的错误); // 抛出异常} 配置好重试之后到了三次就会直接停止这样子就很好的减少了系统资源的消耗 业务的幂等性判断 什么是幂等性 在Java领域幂等性是指同一个请求不管发送多少次执行的结构都是一样的。 比如支付和交易支付成功之后通知交易服务修改状态。在交易服务需要查询订单并判断订单的状态这样子不管同一个订单重复发起多少次请求都不会对业务的结果造成影响。 MQ保证消息的幂等性 MQ中的幂等是说不管消息是否被重复消费都不会对业务造成影响、处理的结果都是一致的。 MQ实现业务幂等性 为每个消息都创建一个唯一的MessageId在操作的时候将其存入数据库然后在进行判断消息是否存在存在就直接跳过业务的处理不存在就继续操作。 Beanpublic MessageConverter messageConverter(){// 1.定义消息转换器Jackson2JsonMessageConverter jackson2JsonMessageConverter new Jackson2JsonMessageConverter();// 2.配置自动创建消息id用于识别不同消息也可以在业务中基于ID判断是否是重复消息jackson2JsonMessageConverter.setCreateMessageIds(true);return jackson2JsonMessageConverter;} 业务实现幂等性 在业务的操作中比如支付和交易服务支付成功之后会通知交易服务修改订单的状态而在交易服务应该做判断判断该订单的状态是否未未支付。如果是未支付就继续处理接下来的业务否则就直接结束。 package com.hmall.trade.listener;import com.hmall.trade.domain.po.Order;
import com.hmall.trade.service.IOrderService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;Component
Slf4j
public class PayStatusListener {AutowiredIOrderService orderService;RabbitListener(bindings QueueBinding(value Queue(value trade.pay.success.queue, durable true),exchange Exchange(value pay.direct, type ExchangeTypes.DIRECT),key pay.success))public void paySuccess(Long orderId) {log.info(支付成功订单号{}, orderId);//查询当前订单 判断幂等性Order order orderService.getById(orderId);//判断状态以及对象是否存在if (order null || order.getStatus() ! 1) {return;}orderService.markOrderPaySuccess(orderId);}
}
