从三个角度入手
1. 生产者消息的可靠性问题
2.Mq可靠性问题
3.消费者消息可靠性问题
1. 生产者消息的可靠性问题
生产者的重连机制
生产者的publisher-return的三种模式类型
/*** 生产者重连机制 发送消息代码* @throws InterruptedException*/@Testvoid testSendMessage2Queue() {String queueName = "simple.queue";String msg = "hello, amqp!";rabbitTemplate.convertAndSend(queueName, msg);System.out.println("msg = " + msg);}
在执行 docker stop mq11 后 关闭 mq
可以发现在尝试
生产者确认机制
两种机制
注意: 通常关闭Publisher Return
生产者对Mq异常消息的回执的接收
ReturnCallback 只能配置一个
/**** @author 16905*/
@Slf4j
@Configuration
public class MqConfirmConfig implements ApplicationContextAware {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {//RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);//生产者 的回调机制 收到 ack 配置回调rabbitTemplate.setReturnsCallback(returned -> log.debug("收到消息的return callback，exchange:{}, key:{}, msg:{}, code:{}, text:{}",returned.getExchange(), returned.getRoutingKey(), returned.getMessage(),returned.getReplyCode(), returned.getReplyText()));}
}
ConfirmCallback 对每个消息都要配置 指定消息的ID
@Test // 生产者确认机制void testConfirmCallback() throws InterruptedException {// 1.创建cdCorrelationData cd = new CorrelationData(UUID.randomUUID().toString());// 2.添加ConfirmCallbackcd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {@Overridepublic void onFailure(Throwable ex) {log.error("消息回调失败", ex);}@Overridepublic void onSuccess(CorrelationData.Confirm result) {log.debug("收到confirm callback回执");if(result.isAck()){// 消息发送成功log.debug("消息发送成功，收到ack");}else{// 消息发送失败log.error("消息发送失败，收到nack， 原因：{}", result.getReason());}}});rabbitTemplate.convertAndSend("hmall.direct123", "red2", "hello", cd);Thread.sleep(2000);}
2.Mq可靠性问题
两种方式
使用消息持久化的方式
mq消息发送会出现page out （相当于刷盘）
/*** 消息发送的持久化*/
@Test
void testPageOut() {Message message = MessageBuilder.withBody("hello".getBytes(StandardCharsets.UTF_8))// 设置消息的持久化.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();for (int i = 0; i < 1000000; i++) {rabbitTemplate.convertAndSend("lazy.queue", message);}
}在发送消息的同时 持久化操作 但不会刷盘
/*** 消息发送的持久化*/
@Test
void testPageOut() {Message message = MessageBuilder.withBody("hello".getBytes(StandardCharsets.UTF_8))// 设置消息的持久化.setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();for (int i = 0; i < 1000000; i++) {rabbitTemplate.convertAndSend("lazy.queue", message);}
}
通过MessageDeliveryMode.PERSISTEN 设置实现消息
设置lazyQueue实现持久化的方式
使用 lazyqueue 的方式 （队列方式）（官方推荐）
响应速度慢一些
、lazyQueue设置方式
、
消息发送结果稳定 保持在60k/s左右
3. 消费者保证消息可靠性
只有这种消息 消息 才会reject
开启消费者确认机制
listener:simple:prefetch: 1acknowledge-mode: auto # 确认机制retry:enabled: true # 开启重试机制
对异常进行处理使用 MessageRecoverer
消息的幂等性问题
方案一 设置消息的唯一ID
方案一 基于业务来判断
业务幂等性的判断的代码实现
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "mark.order.pay.queue", durable = "true"),exchange = @Exchange(name = "pay.topic", type = ExchangeTypes.TOPIC),key = "pay.success"))public void listenOrderPay(Long orderId) {
// 可能会出现并发修改的问题
// 1.查询订单
// Order order = orderService.getById(orderId);
// // 2.判断订单状态是否为未支付
// if(order == null || order.getStatus() != 1){
// // 订单不存在，或者状态异常
// return;
// }
// // 3.如果未支付，标记订单状态为已支付
// orderService.markOrderPaySuccess(orderId);// update order set status = 2 where id = ? AND status = 1 使用这种方式orderService.lambdaUpdate().set(Order::getStatus, 2).set(Order::getPayTime, LocalDateTime.now()).eq(Order::getId, orderId).eq(Order::getStatus, 1).update();}
