SpringBoot 整合RabbitMQ 之延迟队列实验

本文介绍: 实际业务中，例如秒杀系统，秒杀商品成功会有截止时间，这时需要用到Rab bitMQ延迟服务。

第一章 Java线程池技术应用
 第二章 CountDownLatch和Semaphone的应用
 第三章 Spring Cloud 简介
 第四章 Spring Cloud Netflix 之 Eureka
第五章 Spring Cloud Netflix 之 Ribbon
第六章 Spring Cloud 之 OpenFeign
第七章 Spring Cloud 之 GateWay
第八章 Spring Cloud Netflix 之 Hystrix
第九章代码管理gitlab 使用
 第十章 SpringCloud Alibaba 之 Nacos discovery
第十一章 SpringCloud Alibaba 之 Nacos Config
第十二章 Spring Cloud Alibaba 之 Sentinel
第十三章 JWT
第十四章 RabbitMQ应用
 第十五章 RabbitMQ 延迟队列

在这里插入图片描述

实际业务中，例如秒杀系统，秒杀商品成功会有截止时间，这时需要用到Ra b bitMQ延迟服务。

/***********************延迟队列*************************/
//创建立即消费队列
@Bean
public Queue immediateQueue(){
    return new Queue("immediateQueue");
}
//创建立即消费交换机
@Bean
public DirectExchange immediateExchange(){
    return new DirectExchange("immediateExchange");
}
@Bean
public Binding bindingImmediate(@Qualifier("immediateQueue") Queue queue,@Qualifier("immediateExchange") DirectExchange directExchange){
    return BindingBuilder.bind(queue).to(directExchange).with("immediateRoutingKey");
}

//创建延迟队列
@Bean
public Queue delayQueue(){
    Map<String,Object> params = new HashMap<>();
    //死信队列转发的死信转发到立即处理信息的交换机
    params.put("x-dead-letter-exchange","immediateExchange");
    //死信转化携带的routing-key
    params.put("x-dead-letter-routing-key","immediateRoutingKey");
    //设置消息过期时间，单位：毫秒
    params.put("x-message-ttl",60 * 1000);
    return new Queue("delayQueue",true,false,false,params);
}

@Bean
public DirectExchange delayExchange(){
    return new DirectExchange("delayExchange");
}

@Bean
public Binding bindingDelay(@Qualifier("delayQueue") Queue queue,@Qualifier("delayExchange") DirectExchange directExchange){
    return BindingBuilder.bind(queue).to(directExchange).with("delayRoutingKey");
}

@Test
public void sendDelay(){
    this.rabbitTemplate.convertAndSend("delayExchange","delayRoutingKey","Hello world topic");
}

https://github.com/rabbitmq/rabbitmq–d elayed–message–exchange/release s/download/3.11.1/rabbitmq _d elayed _message _exchange-3.11.1.ez
下载解压，到plugins 目录，执行以下的命令：

/**************延迟队列一个单一queue*******************/
@Bean
public Queue delayNewQueue(){
    return new Queue("delayNewQueue");
}
@Bean
public CustomExchange delayNewExchange(){
    Map<String, Object> args = new HashMap<>();
    // 设置类型，可以为fanout、direct、topic
    args.put("x-delayed-type", "direct");
    return new CustomExchange("delayNewExchange","x-delayed-message", true,false,args);
}
@Bean
public Binding bindingNewDelay(@Qualifier("delayNewQueue") Queue queue,@Qualifier("delayNewExchange") CustomExchange customExchange){
    return BindingBuilder.bind(queue).to(customExchange).with("delayNewRoutingKey").noargs();
}

@Test
public void sendDelay() {
    //生产端写完了
    UserInfo userInfo = new UserInfo();
    userInfo.setPassword("13432432");
    userInfo.setUserAccount("kelvin");
    this.rabbitTemplate.convertAndSend("delayNewExchange", "delayNewRoutingKey", userInfo
            , a -> {
                //单位毫秒
                a.getMessageProperties().setDelay(30000);
                return a;
            });
}

消息确认分为两部分：生产确认和消费确认。

生产确认：生产者生产消息后，将消息发送到交换机，触发确认回调；交换机将消息转发到绑定队列，若失败则触发返回回调。
消费确认：默认情况下消息被消费者从队列中获取后即发送确认，不管消费者处理消息时是否失败，不需要额外代码，但是不能保证消息被正确消费。我们增加手动确认，则需要代码中明确进行消息确认。

@Bean
public RabbitTemplate getTemplate(ConnectionFactory connectionFactory){
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    //消息发送到交换器Exchange后触发回调
    template.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            //  可以进行消息入库操作
            log.info("消息唯一标识 correlationData = {}", correlationData);
            log.info("确认结果 ack = {}", ack);
            log.info("失败原因 cause = {}", cause);
        }
    });

    // 配置这个，下面的ReturnCallback 才会起作用
    template.setMandatory(true);
    // 如果消息从交换器发送到对应队列失败时触发（比如 根据发送消息时指定的routingKey找不到队列时会触发）
    template.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
        @Override
        public void returnedMessage(ReturnedMessage returnedMessage) {
            //  可以进行消息入库操作
            log.info("消息主体 message = {}", returnedMessage.getMessage());
            log.info("回复码 replyCode = {}", returnedMessage.getReplyCode());
            log.info("回复描述 replyText = {}", returnedMessage.getReplyText());
            log.info("交换机名字 exchange = {}", returnedMessage.getExchange());
            log.info("路由键 routingKey = {}", returnedMessage.getRoutingKey());

        }
    });
    return template;
}

spring:
  cloud:
    nacos:
      discovery:
        server-addr: localhost:8848
  application:
    name: drp-user-service  #微服务名称
  datasource:
    username: root
    password: root
    url: jdbc:mysql://127.0.0.1:3306/drp
    driver-class-name: com.mysql.cj.jdbc.Driver
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: root
    password: root
    virtual-host: root_vh
    # 确认消息已发送到交换机（Exchange）
    publisher-confirm-type: correlated
    # 确认消息已发送到队列
    publisher-returns: true
    listener:
      simple:
        acknowledge-mode: manual # 开启消息消费手动确认
        retry:
          enabled: true

@RabbitHandler
public void process(UserInfo data, Message message, Channel channel){
    log.info("收到directQueue队列信息：" + data);
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    try {
        //成功消费确认
        channel.basicAck(deliveryTag,true);
        log.info("消费成功确认完毕。。。。。");
    } catch (IOException e) {
        log.error("确认消息时抛出异常 ", e);        
        // 重新确认，成功确认消息
        try {
            Thread.sleep(50);
            channel.basicAck(deliveryTag, true);
        } catch (IOException | InterruptedException e1) {
            log.error("确认消息时抛出异常 ", e);
            // 可以考虑入库
        }
    }catch (Exception e){
        log.error("业务处理失败", e);
        try {
            // 失败确认
            channel.basicNack(deliveryTag, false, false);
        } catch (IOException e1) {
            log.error("消息失败确认失败", e1);
        }
    }
}