怎样做才能提升自己的网站,企业网站建设专业精准丨 鸣远科技,长沙网络营销学校,店铺推广渠道有哪些方式目录 一、消费者确认
二、失败重试机制
2.1、失败处理策略
三、业务幂等性
3.1、唯一消息ID 3.2、业务判断
3.3、兜底方案 一、消费者确认
RabbitMQ提供了消费者确认机制#xff08;Consumer Acknowledgement#xff09;。即#xff1a;当消费者处理消息结束后#x…目录 一、消费者确认
二、失败重试机制
2.1、失败处理策略
三、业务幂等性
3.1、唯一消息ID 3.2、业务判断
3.3、兜底方案 一、消费者确认
RabbitMQ提供了消费者确认机制Consumer Acknowledgement。即当消费者处理消息结束后应该向RabbitMQ发送一个回执告知RabbitMQ自己消息处理状态。回执有三种可选值
ack成功处理消息RabbitMQ从队列中删除该消息nack消息处理失败RabbitMQ需要再次投递消息reject消息处理失败并拒绝该消息RabbitMQ从队列中删除该消息
一般reject方式用的较少除非是消息格式有问题那就是开发问题了。因此大多数情况下我们需要将消息处理的代码通过try catch机制捕获消息处理成功时返回ack处理失败时返回nack.
由于消息回执的处理代码比较统一因此SpringAMQP帮我们实现了消息确认。并允许我们通过配置文件设置ACK处理方式有三种模式
none不处理。即消息投递给消费者后立刻ack消息会立刻从MQ删除。非常不安全不建议使用manual手动模式。需要自己在业务代码中调用api发送ack或reject存在业务入侵但更灵活auto自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强当业务正常执行时则自动返回ack. 当业务出现异常时根据异常判断返回不同结果 如果是业务异常会自动返回nack如果是消息处理或校验异常自动返回reject; 通过下面的配置可以修改SpringAMQP的ACK处理方式
spring:rabbitmq:host: 192.168.200.129 # 你的虚拟机IPport: 5672 # 端口virtual-host: / # 虚拟主机username: admin # 用户名password: 123 # 密码listener:simple:prefetch: 1 # 每次只能获取一条消息处理完成才能获取下一个消息acknowledge-mode: none当使用none模式时
生产者发送一条消息 消费者接受消息时抛异常 测试可以发现当消息处理发生异常时消息依然被RabbitMQ删除了。
把确认机制修改为auto
spring:rabbitmq:host: 192.168.200.129 # 你的虚拟机IPport: 5672 # 端口virtual-host: / # 虚拟主机username: admin # 用户名password: 123 # 密码listener:simple:prefetch: 1 # 每次只能获取一条消息处理完成才能获取下一个消息acknowledge-mode: auto 再次发送消息 在异常位置打断点再次发送消息程序卡在断点时可以发现此时消息状态为unacked未确定状态 当我们把配置改为auto时消息处理失败后会回到RabbitMQ并重新投递到消费者。 二、失败重试机制
当消费者出现异常后消息会不断requeue重入队到队列再重新发送给消费者。如果消费者再次执行依然出错消息会再次requeue到队列再次投递直到消息处理成功为止。 极端情况就是消费者一直无法执行成功那么消息requeue就会无限循环导致mq的消息处理飙升带来不必要的压力。
为了应对上述情况Spring又提供了消费者失败重试机制在消费者出现异常时利用本地重试而不是无限制的requeue到mq队列。
修改consumer服务的application.yml文件添加内容
spring:rabbitmq:host: 192.168.200.129 # 你的虚拟机IPport: 5672 # 端口virtual-host: / # 虚拟主机username: admin # 用户名password: 123 # 密码listener:simple:prefetch: 1 # 每次只能获取一条消息处理完成才能获取下一个消息acknowledge-mode: auto #消息确认retry:enabled: true # 开启消费者失败重试initial-interval: 1000ms # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数下次等待时长 multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态false有状态。如果业务中包含事务这里改为false 重启consumer服务重复之前的测试。可以发现
消费者在失败后消息没有重新回到MQ无限重新投递而是在本地重试了3次
结论
开启本地重试时消息处理过程中抛出异常不会requeue到队列而是在消费者本地重试重试达到最大次数后Spring会返回reject消息会被丢弃
2.1、失败处理策略
在之前的测试中本地测试达到最大重试次数后消息会被丢弃。这在某些对于消息可靠性要求较高的业务场景下显然不太合适了
因此Spring允许我们自定义重试次数耗尽后的消息处理策略这个策略是由MessageRecovery接口来定义的它有3个不同实现
RejectAndDontRequeueRecoverer重试耗尽后直接reject丢弃消息。默认就是这种方式ImmediateRequeueMessageRecoverer重试耗尽后返回nack消息重新入队RepublishMessageRecoverer重试耗尽后将失败消息投递到指定的交换机
在消费者里创建一个异常消息配置类
Configuration
Slf4j
//当配置文件中spring.rabbitmq.listener.simple.retry.enabled 属性为ture时配置类才生效
ConditionalOnProperty(name spring.rabbitmq.listener.simple.retry.enabled, havingValue true)
public class ErrorMessageConfig {//消息处理失败交换机Beanpublic DirectExchange errorMessageExchange(){return new DirectExchange(error.direct);}//消息处理失败队列Beanpublic Queue errorQueue(){return new Queue(error.queue, true);}//绑定关系Beanpublic Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with(error);}//定义一个RepublishMessageRecoverer关联队列和交换机Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){log.error(加载RepublishMessageRecoverer);return new RepublishMessageRecoverer(rabbitTemplate, error.direct, error);}
} 当接收消息出现异常时会创建error.queue队列 可以查看到异常信息 三、业务幂等性
何为幂等性 幂等是一个数学概念用函数表达式来描述是这样的f(x) f(f(x))例如求绝对值函数。 在程序开发中则是指同一个业务执行一次或多次对业务状态的影响是一致的。例如
根据id删除数据查询数据新增数据
但数据的更新往往不是幂等的如果重复执行可能造成不一样的后果。比如
取消订单恢复库存的业务。如果多次恢复就会出现库存重复增加的情况退款业务。重复退款对商家而言会有经济损失。
所以我们要尽可能避免业务被重复执行。 然而在实际业务场景中由于意外经常会出现业务被重复执行的情况例如
页面卡顿时频繁刷新导致表单重复提交服务间调用的重试MQ消息的重复投递
我们在用户支付成功后会发送MQ消息到交易服务修改订单状态为已支付就可能出现消息重复投递的情况。如果消费者不做判断很有可能导致消息被消费多次出现业务故障。 举例
假如用户刚刚支付完成并且投递消息到交易服务交易服务更改订单为已支付状态。由于某种原因例如网络故障导致生产者没有得到确认隔了一段时间后重新投递给交易服务。但是在新投递的消息被消费之前用户选择了退款将订单状态改为了已退款状态。退款完成后新投递的消息才被消费那么订单状态会被再次改为已支付。业务异常。
因此我们必须想办法保证消息处理的幂等性。这里给出两种方案
唯一消息ID业务状态判断
3.1、唯一消息ID
每一条消息都生成一个唯一的id与消息一起投递给消费者。消费者接收到消息后处理自己的业务业务处理成功后将消息ID保存到数据库如果下次又收到相同消息去数据库查询判断是否存在存在则为重复消息放弃处理。
SpringAMQP的MessageConverter自带了MessageID的功能我们只要开启这个功能即可。 以Jackson的消息转换器为例
在生产者和消费者的启动类里加一个配置
Bean
public MessageConverter messageConverter(){// 1.定义消息转换器Jackson2JsonMessageConverter jjmc new Jackson2JsonMessageConverter();// 2.配置自动创建消息id用于识别不同消息也可以在业务中基于ID判断是否是重复消息jjmc.setCreateMessageIds(true);return jjmc;
} 测试生产者发送消息会产生一个id 3.2、业务判断
业务判断就是基于业务本身的逻辑或状态来判断是否是重复的请求或消息不同的业务场景判断的思路也不一样。 例如我们当前案例中处理消息的业务逻辑是把订单状态从未支付修改为已支付。因此我们就可以在执行业务时判断订单状态是否是未支付如果不是则证明订单已经被处理过无需重复处理。 相比较而言消息ID的方案需要改造原有的数据库所以我更推荐使用业务判断的方案。
以支付修改订单的业务为例我们需要修改OrderServiceImpl中的markOrderPaySuccess方法 Overridepublic void markOrderPaySuccess(Long orderId) {// 1.查询订单Order old getById(orderId);// 2.判断订单状态if (old null || old.getStatus() ! 1) {// 订单不存在或者订单状态不是1放弃处理return;}// 3.尝试更新订单Order order new Order();order.setId(orderId);order.setStatus(2);order.setPayTime(LocalDateTime.now());updateById(order);}
上述代码逻辑上符合了幂等判断的需求但是由于判断和更新是两步动作因此在极小概率下可能存在线程安全问题。
Override
public void markOrderPaySuccess(Long orderId) {// UPDATE order SET status ? , pay_time ? WHERE id ? AND status 1lambdaUpdate().set(Order::getStatus, 2).set(Order::getPayTime, LocalDateTime.now()).eq(Order::getId, orderId).eq(Order::getStatus, 1).update();
}
注意看上述代码等同于这样的SQL语句
UPDATE order SET status ? , pay_time ? WHERE id ? AND status 1我们在where条件中除了判断id以外还加上了status必须为1的条件。如果条件不符说明订单已支付则SQL匹配不到数据根本不会执行。
3.3、兜底方案
我们可以在交易服务设置定时任务定期查询订单支付状态。这样即便MQ通知失败还可以利用定时任务作为兜底方案确保订单支付状态的最终一致性。
