一、消费者确认

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;

192.168.200.129

每次只能获取一条消息处理完成才能获取下一个消息acknowledge-mode:

none当使用none模式时

测试可以发现当消息处理发生异常时消息依然被RabbitMQ删除了。

把确认机制修改为auto

每次只能获取一条消息处理完成才能获取下一个消息acknowledge-mode:

auto

在异常位置打断点再次发送消息程序卡在断点时可以发现此时消息状态为unacked未确定状态

当我们把配置改为auto时消息处理失败后会回到RabbitMQ并重新投递到消费者。

二、失败重试机制

当消费者出现异常后消息会不断requeue重入队到队列再重新发送给消费者。

如果消费者再次执行依然出错消息会再次requeue到队列再次投递直到消息处理成功为止。

极端情况就是消费者一直无法执行成功那么消息requeue就会无限循环导致mq的消息处理飙升带来不必要的压力。

为了应对上述情况Spring又提供了消费者失败重试机制在消费者出现异常时利用本地重试而不是无限制的requeue到mq队列。

修改consumer服务的application.yml文件添加内容

192.168.200.129

每次只能获取一条消息处理完成才能获取下一个消息acknowledge-mode:

auto

true无状态false有状态。

如果业务中包含事务这里改为false

消费者在失败后消息没有重新回到MQ无限重新投递而是在本地重试了3次

开启本地重试时消息处理过程中抛出异常不会requeue到队列而是在消费者本地重试重试达到最大次数后Spring会返回reject消息会被丢弃

2.1、失败处理策略

在之前的测试中本地测试达到最大重试次数后消息会被丢弃。

这在某些对于消息可靠性要求较高的业务场景下显然不太合适了

因此Spring允许我们自定义重试次数耗尽后的消息处理策略这个策略是由MessageRecovery接口来定义的它有3个不同实现

RejectAndDontRequeueRecoverer重试耗尽后直接reject丢弃消息。

默认就是这种方式ImmediateRequeueMessageRecoverer重试耗尽后返回nack消息重新入队RepublishMessageRecoverer重试耗尽后将失败消息投递到指定的交换机

在消费者里创建一个异常消息配置类

//当配置文件中spring.rabbitmq.listener.simple.retry.enabled

属性为ture时配置类才生效

spring.rabbitmq.listener.simple.retry.enabled,

havingValue

DirectExchange(error.direct);}//消息处理失败队列Beanpublic

Queue

BindingBuilder.bind(errorQueue).to(errorMessageExchange).with(error);}//定义一个RepublishMessageRecoverer关联队列和交换机Beanpublic

MessageRecoverer

republishMessageRecoverer(RabbitTemplate

rabbitTemplate){log.error(加载RepublishMessageRecoverer);return

new

RepublishMessageRecoverer(rabbitTemplate,

error.direct,

在程序开发中则是指同一个业务执行一次或多次对业务状态的影响是一致的。

例如

根据id删除数据查询数据新增数据

但数据的更新往往不是幂等的如果重复执行可能造成不一样的后果。

比如

取消订单恢复库存的业务。

如果多次恢复就会出现库存重复增加的情况退款业务。

重复退款对商家而言会有经济损失。

然而在实际业务场景中由于意外经常会出现业务被重复执行的情况例如

页面卡顿时频繁刷新导致表单重复提交服务间调用的重试MQ消息的重复投递

我们在用户支付成功后会发送MQ消息到交易服务修改订单状态为已支付就可能出现消息重复投递的情况。

如果消费者不做判断很有可能导致消息被消费多次出现业务故障。

假如用户刚刚支付完成并且投递消息到交易服务交易服务更改订单为已支付状态。

由于某种原因例如网络故障导致生产者没有得到确认隔了一段时间后重新投递给交易服务。

但是在新投递的消息被消费之前用户选择了退款将订单状态改为了已退款状态。

退款完成后新投递的消息才被消费那么订单状态会被再次改为已支付。

业务异常。

唯一消息ID业务状态判断

每一条消息都生成一个唯一的id与消息一起投递给消费者。

消费者接收到消息后处理自己的业务业务处理成功后将消息ID保存到数据库如果下次又收到相同消息去数据库查询判断是否存在存在则为重复消息放弃处理。

SpringAMQP的MessageConverter自带了MessageID的功能我们只要开启这个功能即可。

Bean

1.定义消息转换器Jackson2JsonMessageConverter

jjmc

Jackson2JsonMessageConverter();//

2.配置自动创建消息id用于识别不同消息也可以在业务中基于ID判断是否是重复消息jjmc.setCreateMessageIds(true);return

jjmc;

业务判断就是基于业务本身的逻辑或状态来判断是否是重复的请求或消息不同的业务场景判断的思路也不一样。

例如我们当前案例中处理消息的业务逻辑是把订单状态从未支付修改为已支付。

因此我们就可以在执行业务时判断订单状态是否是未支付如果不是则证明订单已经被处理过无需重复处理。

相比较而言消息ID的方案需要改造原有的数据库所以我更推荐使用业务判断的方案。

以支付修改订单的业务为例我们需要修改OrderServiceImpl中的markOrderPaySuccess方法

Overridepublic

Order();order.setId(orderId);order.setStatus(2);order.setPayTime(LocalDateTime.now());updateById(order);}

上述代码逻辑上符合了幂等判断的需求但是由于判断和更新是两步动作因此在极小概率下可能存在线程安全问题。

Override

1lambdaUpdate().set(Order::getStatus,

LocalDateTime.now()).eq(Order::getId,

1).update();

1我们在where条件中除了判断id以外还加上了status必须为1的条件。

如果条件不符说明订单已支付则SQL匹配不到数据根本不会执行。

3.3、兜底方案

我们可以在交易服务设置定时任务定期查询订单支付状态。

这样即便MQ通知失败还可以利用定时任务作为兜底方案确保订单支付状态的最终一致性。