投稿
  1. 首页
  2. 随笔

消息队列 — RabbitMQ07

低通滤波器电路 • 2022-12-17 • 随笔 • 阅读 76

消息队列 — RabbitMQ07,第1张

消息队列 — RabbitMQ07 7 延迟队列 7.1 概念

延时队列内部是有序的，最重要的特性就体现在它的延时属性上，延时队列中的元素是希望在指定时间到了以后或之前取出和处理，延时队列是用来存放需要在指定时间被处理的元素的队列。

7.2 使用场景

  1. 订单在十分钟之内未支付则自动取消

  2. 新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒

  3. 用户注册成功后，如果三天内没有登陆则进行短信提醒

  4. 用户发起退款，如果三天内没有得到处理则通知相关运营人员

  5. 预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议

7.3 RabbitMQ中的TTL

TTL是什么？

TTL是RabbitMQ中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间，单位是毫秒。换句话说，如果一条消息设置了TTL属性或者进入了设置TTL 属性的队列，那么这条消息如果在TTL 设置的时间内没有被消费，则会成为"死信"。如果同时配置了队列的TTL和消息的TTL，那么较小的那个值将会被使用，有两种方式设置TTL。

  1. 消息设置TTL(msgProperties.setExpiration(ttlTime))

  2. 队列设置TTL(x-message-tt)

如果设置了队列的 TTL 属性，那么一旦消息过期，就会被队列丢弃(如果配置了死信队列被丢到死信队列中)

而第二种方式，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间；

另外，还需要注意的一点是，如果不设置 TTL，表示消息永远不会过期，如果将 TTL 设置为 0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

7.4 整合 Springboot 7.4.1 增加Pom依赖 


  4.0.0

  com.zjw
  hello-rabbitmq
  1.0-SNAPSHOT

  
    org.springframework.boot
    spring-boot-starter-parent
    2.4.5
  

  
    1.8
    5.14.0
    2.11.0
  

  
    
    
      org.springframework.boot
      spring-boot-starter-amqp
    
    
      org.springframework.boot
      spring-boot-starter-web
    
    
      org.springframework.boot
      spring-boot-starter-test
      test
    
    
      com.alibaba
      fastjson
      1.2.47
    
    
      org.projectlombok
      lombok
    
    
    
      io.springfox
      springfox-swagger2
      2.9.2
    
    
      io.springfox
      springfox-swagger-ui
      2.9.2
    
    
    
      org.springframework.amqp
      spring-rabbit-test
      test
    
    
      commons-io
      commons-io
      ${commons-io.version}
    
  

  
    
      
        org.apache.maven.plugins
        maven-compiler-plugin
        
          ${java.version}
          ${java.version}
7.4.3 修改配置文件 
spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: admin
    password: admin
7.4.4 添加Swagger配置类 
package com.zjw.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import springfox.documentation.builders.ApiInfoBuilder;
import springfox.documentation.service.ApiInfo;
import springfox.documentation.service.Contact;
import springfox.documentation.spi.documentationType;
import springfox.documentation.spring.web.plugins.Docket;
import springfox.documentation.swagger2.annotations.EnableSwagger2;


@Configuration
@EnableSwagger2
public class SwaggerConfig {

  @Bean
  public Docket webApiConfig() {
    return new Docket(documentationType.SWAGGER_2)
        .groupName("webApi")
        .apiInfo(webApiInfo())
        .select()
        .build();
  }

  private ApiInfo webApiInfo() {
    return new ApiInfoBuilder()
        .title("rabbitmq 接口文档")
        .description("本文档描述了 rabbitmq 微服务接口定义")
        .version("v1.0")
        .contact(new Contact("chdaring", "https://blog.csdn.net/co_zjw", "cn.zjwblog@gmail.com"))
        .build();
  }
}
7.5 队列 TTL 7.5.1 代码架构图

7.5.2 配置文件类代码 
package com.zjw.config;

import java.util.HashMap;
import java.util.Map;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class TtlQueueConfig {

  public static final String X_EXCHANGE = "X";
  public static final String QUEUE_A = "QA";
  public static final String QUEUE_B = "QB";
  public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
  public static final String DEAD_LETTER_QUEUE = "QD";

  // 声明 xExchange
  @Bean("xExchange")
  public DirectExchange xExchange() {
    return new DirectExchange(X_EXCHANGE);
  }

  // 声明 xExchange
  @Bean("yExchange")
  public DirectExchange yExchange() {
    return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
  }

  //声明队列 A ttl 为 10s 并绑定到对应的死信交换机
  @Bean("queueA")
  public Queue queueA() {
    Map args = new HashMap<>(3);
    //声明当前队列绑定的死信交换机
    args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
    //声明当前队列的死信路由 key
    args.put("x-dead-letter-routing-key", "YD");
    //声明队列的 TTL
    args.put("x-message-ttl", 10000);
    return QueueBuilder.durable(QUEUE_A).withArguments(args).build();
  }

  // 声明队列 A 绑定 X 交换机
  @Bean
  public Binding queueaBindingX(@Qualifier("queueA") Queue queueA,
      @Qualifier("xExchange") DirectExchange xExchange) {
    return BindingBuilder.bind(queueA).to(xExchange).with("XA");
  }

  //声明队列 B ttl 为 40s 并绑定到对应的死信交换机
  @Bean("queueB")
  public Queue queueB() {
    Map args = new HashMap<>(3);
    //声明当前队列绑定的死信交换机
    args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
    //声明当前队列的死信路由 key
    args.put("x-dead-letter-routing-key", "YD");
    //声明队列的 TTL
    args.put("x-message-ttl", 40000);
    return QueueBuilder.durable(QUEUE_B).withArguments(args).build();
  }

  //声明队列 B 绑定 X 交换机
  @Bean
  public Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,
      @Qualifier("xExchange") DirectExchange xExchange) {
    return BindingBuilder.bind(queue1B).to(xExchange).with("XB");
  }

  //声明死信队列 QD
  @Bean("queueD")
  public Queue queueD() {
    return new Queue(DEAD_LETTER_QUEUE);
  }

  //声明死信队列 QD 绑定关系
  @Bean
  public Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,
      @Qualifier("yExchange") DirectExchange yExchange) {
    return BindingBuilder.bind(queueD).to(yExchange).with("YD");
  }
}
7.5.3 消息生产者代码 
package com.zjw.controller;

import java.text.SimpleDateFormat;
import java.util.Date;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@Slf4j
@RequestMapping("ttl")
@RestController
public class SendMsgController {
  private final SimpleDateFormat SDF = new SimpleDateFormat("yyy-MM-dd HH:mm:ss.SSS");

  @Resource
  private RabbitTemplate rabbitTemplate;

  @GetMapping("sendMsg/{message}")
  public void sendMsg(@PathVariable String message) {
    log.info("当前时间：{},发送一条信息给两个 TTL 队列:{}", SDF.format(new Date()), message);
    rabbitTemplate.convertAndSend("X", "XA", "消息来自 ttl 为 10S 的队列: " + message);
    rabbitTemplate.convertAndSend("X", "XB", "消息来自 ttl 为 40S 的队列: " + message);
  }

}
7.5.4 消息消费者代码 
package com.zjw.component;

import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;


@Slf4j
@Component
public class DeadLetterQueueConsumer {
  private final SimpleDateFormat SDF = new SimpleDateFormat("yyy-MM-dd HH:mm:ss.SSS");

  @RabbitListener(queues = "QD")
  public void receiveD(Message message, Channel channel) throws IOException {
    String msg = new String(message.getBody());
    log.info("当前时间：{},收到死信队列信息{}", SDF.format(new Date()), msg);
  }

}

缺点：只能进行固定的延时，没办法指定延时时长

7.6 延时队列优化 7.6.1 代码架构图

7.6.2 配置文件类代码 
package com.zjw.config;

import java.util.HashMap;
import java.util.Map;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;


@Component
public class MsgTtlQueueConfig {

  public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
  public static final String QUEUE_C = "QC";

  //声明队列 C 死信交换机
  @Bean("queueC")
  public Queue queueB() {
    Map args = new HashMap<>(3);
    // 声明当前队列绑定的死信交换机
    args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
    // 声明当前队列的死信路由 key
    args.put("x-dead-letter-routing-key", "YD");
    // 没有声明 TTL 属性
    return QueueBuilder.durable(QUEUE_C).withArguments(args).build();
  }

  // 声明队列 B 绑定 X 交换机
  @Bean
  public Binding queuecBindingX(@Qualifier("queueC") Queue queueC,
      @Qualifier("xExchange") DirectExchange xExchange) {
    return BindingBuilder.bind(queueC).to(xExchange).with("XC");
  }
}
7.6.3 消息生产者代码 
package com.zjw.controller;

import java.text.SimpleDateFormat;
import java.util.Date;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@Slf4j
@RequestMapping("ttl")
@RestController
public class SendMsgController {
  private final SimpleDateFormat SDF = new SimpleDateFormat("yyy-MM-dd HH:mm:ss.SSS");
  @Resource
  private RabbitTemplate rabbitTemplate;

  @GetMapping("sendExpirationMsg/{message}/{ttlTime}")
  public void sendMsg(@PathVariable String message, @PathVariable String ttlTime) {
    rabbitTemplate.convertAndSend("X", "XC", message, correlationData -> {
      correlationData.getMessageProperties().setExpiration(ttlTime);
      return correlationData;
    });
    log.info("当前时间：{},发送一条时长{}毫秒 TTL 信息给队列 C:{}", new Date(), ttlTime, message);
  }
}

缺点：延时只能顺序延时，例如：后面的延时时间短也没办法提前被加入到延时队列中.

7.7 Rabbitmq 插件实现延迟队列 7.7.1 安装延时队列插件

在官网上下载 https://www.rabbitmq.com/community-plugins.html，下载rabbitmq_delayed_message_exchange 插件，然后解压放置到 RabbitMQ 的插件目录。进入 RabbitMQ 的安装目录下的 plgins 目录，执行rabbitmq-plugins enable rabbitmq_delayed_message_exchange使插件生效，然后重启服务。

7.7.2 配置文件类代码 
package com.zjw.config;

import java.util.HashMap;
import java.util.Map;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class DelayedQueueConfig {

  public static final String DELAYED_QUEUE_NAME = "delayed.queue";
  public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
  public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";

  @Bean
  public Queue delayedQueue() {
    return new Queue(DELAYED_QUEUE_NAME);
  }

  //自定义交换机 我们在这里定义的是一个延迟交换机
  @Bean
  public CustomExchange delayedExchange() {
    Map args = new HashMap<>();
    //自定义交换机的类型
    args.put("x-delayed-type", "direct");
    return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, args);
  }

  @Bean
  public Binding bindingDelayedQueue(
      @Qualifier("delayedQueue") Queue queue,
      @Qualifier("delayedExchange"
      ) CustomExchange delayedExchange) {
    return BindingBuilder.bind(queue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
  }
}
7.7.3 消息生产者代码 
package com.zjw.controller;

import java.text.SimpleDateFormat;
import java.util.Date;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@Slf4j
@RequestMapping("ttl")
@RestController
public class SendMsgController {

  public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
  public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
  private final SimpleDateFormat SDF = new SimpleDateFormat("yyy-MM-dd HH:mm:ss.SSS");
  @Resource
  private RabbitTemplate rabbitTemplate;

  @GetMapping("sendDelayMsg/{message}/{delayTime}")
  public void sendMsg(@PathVariable String message, @PathVariable Integer delayTime) {
    rabbitTemplate.convertAndSend(
        DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, message,
        correlationData -> {
          correlationData.getMessageProperties().setDelay(delayTime);
          return correlationData;
        });
    log.info("当前时间: {}, 延迟时间: {}, 发送消息: {}", SDF.format(new Date()), delayTime, message);
  }

}
7.7.4 消息消费者代码 
package com.zjw.component;

import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;


@Slf4j
@Component
public class DeadLetterQueueConsumer {
  private final SimpleDateFormat SDF = new SimpleDateFormat("yyy-MM-dd HH:mm:ss.SSS");
  private final String DELAYED_QUEUE_NAME = "delayed.queue";

  @RabbitListener(queues = DELAYED_QUEUE_NAME)
  public void receiveDelayedQueue(Message
      message){String msg = new
      String(message.getBody());
    log.info("当前时间: {},收到延时队列的消息: {}", SDF.format(new Date()), msg);
  }
}
7.8 总结

延时队列在需要延时处理的场景下非常有用，使用 RabbitMQ 来实现延时队列可以很好的利用RabbitMQ 的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过 RabbitMQ 集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。

当然，延时队列还有很多其它选择，比如利用 Java 的 DelayQueue，利用 Redis 的 zset，利用 Quartz或者利用 kafka 的时间轮，这些方式各有特点,看需要适用的场景

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/zaji/5683673.html

赞 (0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
低通滤波器电路 低通滤波器电路 一级用户组
0 0
上一篇 2022-12-17
下一篇 2022-12-17

发表评论

登录后才能评论

评论列表（0条）

保存