做网站好的网络公司_web前端开发怎么学_线上宣传推广方式_宁波网站建设网站排名优化

从三个角度入手  
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();}