阅读量:0
失败处理策略
在高级一中，我们设置了消息重连机制设置了次数，可是对于要求可靠性较高的消息，直接从删除是比较不好的选择，因此我们就有了失败处理的策略
这个策略是由MessageRecovery接口来定义的，它有3个不同实现：
RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认就是这种方式ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机
最好的就是第三种我们设置一个专门存放错误消息的消息队列，这样就后续由人工集中处理。
1）定义处理失败消息的交换机和队列（我们就基于构造函数来创建）
@Bean public DirectExchange errorMessageExchange(){ return new DirectExchange("error.direct"); } @Bean public Queue errorQueue(){ return new Queue("error.queue", true); } @Bean public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){ return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error"); }2）定义一个RepublishMessageRecoverer，关联队列和交换机（此时重复次数达到我们就会发送消息到这个交换机）
@Bean public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){ return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error"); }业务幂等性
在消息队列发送消息给消费者消费时，消费者已完成了自己的逻辑，可是在给mq返回ask成功消息时，由于网络不稳定或者其他原因出现异常，导致消费者连接不上mq，此时mq会认为消费者这边出现了宕机，mq就会把消息重新放入消息队列中，当消费者连接上了mq，消息队列又会再次投递同一条消息给消费者进行处理，此时就会出现业务不幂等。
幂等是一个数学概念，用函数表达式来描述是这样的：f(x) = f(f(x))，例如求绝对值函数。
因此，我们就产生了两种方案用来解决业务的幂等性问题
唯一消息ID
业务状态判断
唯一消息ID
每一条消息都生成一个唯一的id，与消息一起投递给消费者。
消费者接收到消息后处理自己的业务，业务处理成功后将消息ID保存到数据库
如果下次又收到相同消息，去数据库查询判断是否存在，存在则为重复消息放弃处理。
我们可以在自己配置的jackson消息转换其中增加配置（此时每一条消息都会有自己的id）：
@Bean public MessageConverter messageConverter(){ // 1.定义消息转换器 Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter(); // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息 jjmc.setCreateMessageIds(true); return jjmc; }有了id我们只需要创建存储id与消息的表字段
每次监听到消息时，我们都去往这张表查有没有这个id字段的消息存在，有的话我们就不做处理，没有的话我们就把这条消息插入到表中，并处理这条消息。
这种方案带来的问题就是：
1.我们需要到单独去维护一张表，影响效率，
2.在业务代码中不应该有其他业务操作，此时有了代码入侵，
3.该逻辑还要去查询操作数据库，会影响业务本身的效率
业务判断
业务判断就是基于业务本身的逻辑或状态来判断是否是重复的请求或消息，不同的业务场景判断的思路也不一样。
例：
// 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);我们可以合并这两个操作
lambdaUpdate() .set(Order::getStatus, 2) .set(Order::getPayTime, LocalDateTime.now()) .eq(Order::getId, orderId) .eq(Order::getStatus, 1) .update();兜底方案
既然MQ通知不一定发送到交易服务，那么交易服务就必须自己主动去查询支付状态。这样即便支付服务的MQ通知失败，我们依然能通过主动查询来保证订单状态的一致。
查询的时机我们无法确定，此时，通常我们采取的措施就是利用定时任务定期查询，例如每隔20秒就查询一次，并判断支付状态。如果发现订单已经支付，则立刻更新订单状态为已支付即可。
支付服务与交易服务之间的订单状态一致性是如何保证的？
首先，支付服务会正在用户支付成功以后利用MQ消息通知交易服务，完成订单状态同步。
其次，为了保证MQ消息的可靠性，我们采用了生产者确认机制、消费者确认、消费者失败重试等策略，确保消息投递的可靠性
最后，我们还在交易服务设置了定时任务，定期查询订单支付状态。这样即便MQ通知失败，还可以利用定时任务作为兜底方案，确保订单支付状态的最终一致性。
延迟消息
对于超过一定时间未支付的订单，应该立刻取消订单并释放占用的库存。
如何才能准确的实现在下单后第30分钟去检查支付状态呢？
像这种在一段时间以后才执行的任务，我们称之为延迟任务，而要实现延迟任务，最简单的方案就是利用MQ的延迟消息了。
在RabbitMQ中实现延迟消息也有两种方案：
死信交换机+TTL
延迟消息插件
死信交换机和延迟消息
死信交换机
死信（dead letter）：
消费者使用
basic.reject或basic.nack声明消费失败，并且消息的requeue参数设置为false消息是一个过期消息，超时无人消费
要投递的队列消息满了，无法投递
如果一个队列中的消息已经成为死信，并且这个队列通过dead-letter-exchange属性指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机就称为死信交换机（Dead Letter Exchange）。而此时加入有队列与死信交换机绑定，则最终死信就会被投递到这个队列中。
死信交换机有什么作用呢？
收集那些因处理失败而被拒绝的消息
收集那些因队列满了而被拒绝的消息
收集因TTL（有效期）到期的消息
延迟消息
投递到ttl.queue之后，由于没有消费者，因此消息无人消费。5秒之后，消息的有效期到期，成为死信：
死信被再次投递到死信交换机hmall.direct，并沿用之前的RoutingKey，也就是blue：
由于direct.queue1与hmall.direct绑定的key是blue，因此最终消息被成功路由到direct.queue1，如果此时有消费者与direct.queue1绑定， 也就能成功消费消息了。但此时已经是5秒钟以后了：
publisher发送了一条消息，但最终consumer在5秒后才收到消息。我们成功实现了延迟消息。
注意：
RabbitMQ的消息过期是基于追溯方式来实现的，也就是说当一个消息的TTL到期以后不一定会被移除或投递到死信交换机，而是在消息恰好处于队首时才会被处理。
当队列中消息堆积很多的时候，过期消息可能不会被按时处理，因此你设置的TTL时间不一定准确。
DelayExchange插件
基于死信队列虽然可以实现延迟消息，但是太麻烦了。因此RabbitMQ社区提供了一个延迟消息插件来实现相同的效果。
官方文档说明：
https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq
下载
插件下载地址：
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange
安装
因为我们是基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷。
docker volume inspect mq-plugins数据卷挂载的地址："Mountpoint": "/var/lib/docker/volumes/mq-plugins/_data"
我们上传插件到该目录下。
接下来执行命令，安装插件：
docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange
此时插件就安装好了
声明延迟交换机
基于注解方式：
@RabbitListener(bindings = @QueueBinding( value = @Queue(name = "delay.queue", durable = "true"), exchange = @Exchange(name = "delay.direct", delayed = "true"), key = "delay" )) public void listenDelayMessage(String msg){ log.info("接收到delay.queue的延迟消息：{}", msg); }发送延迟消息
必须通过x-delay属性设定延迟时间：
@Test void testPublisherDelayMessage() { // 1.创建消息 String message = "hello, delayed message"; // 2.发送消息，利用消息后置处理器添加消息头 rabbitTemplate.convertAndSend("delay.direct", "delay", message, new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { // 添加延迟消息属性 message.getMessageProperties().setDelay(5000); return message; } }); }注意：
延迟消息插件内部会维护一个本地数据库表，同时使用Elang Timers功能实现计时。如果消息的延迟时间设置较长，可能会导致堆积的延迟消息非常多，会带来较大的CPU开销，同时延迟消息的时间会存在误差。
因此，不建议设置延迟时间过长的延迟消息。
