当前位置： 首页 > article >正文

RabbitMQ-TTL机制

article 2025/1/5 9:08:48

RabbitMQ的TTL机制

在这里插入图片描述

在支付类的场景中，创建订单成功后，一般会给30分钟左右的等待时间，如果在这段时间内用户没有支付，则默认订单取消。

那如何实现呢？

8.1 定期轮询（数据库等）

用户下单成功，将订单信息放入数据库，同时将支付状态放入数据库，用户付款更改数据库状态。定期轮询数据库支付状态，如果超过30分钟就将该订单取消。

优点：设计实现简单

缺点：需要对数据库进行大量的IO操作，效率低下。

8.2 定时操作-Timer

  public void timerTask() throws Exception {
    Timer timer = new Timer();

    TimerTask task =
        new TimerTask() {
          @Override
          public void run() {
            LocalDateTime outTime = LocalDateTime.now();
            System.out.println("用户没有付款，交易取消:" + outTime);
            timer.cancel();
          }
        };

    System.out.println("等待用户付款:" + LocalDateTime.now());
    // 在20秒后执行定时调用
    timer.schedule(task, 5 * 1000);

    Thread.sleep(7000);
  }

缺点：

  1. timers没有持久化机制。
  2. timers不灵活（只可以设置开始时间与重复间隔，但对于等待支付的场景是够用的）
  3. timers不能够利用池，一个timer一个线程。
  4. times没有真正的管理计划。

8.3 使用ScheduledExecutorService

  public void scheduleExecute() throws Exception {
    ThreadFactory factory = Executors.defaultThreadFactory();
    ScheduledExecutorService service = new ScheduledThreadPoolExecutor(10, factory);

    System.out.println("等待用户付款:" + LocalDateTime.now());

    service.schedule(
        new Runnable() {
          @Override
          public void run() {
            LocalDateTime outTime = LocalDateTime.now();
            System.out.println("用户没有付款，交易取消:" + outTime);
          }
        },
        // 等待5秒
        5,
        TimeUnit.SECONDS);

    Thread.sleep(7000);
  }

优点：可以多线程执行，一定程序上避免任何任务间的想到影响，单个任务异常不影响其他任务。

缺点：在高并情况下，不建议使用定时任务去做，因为太浪费服务器性能。不推荐。

8.4 RabbitMQ的TTL机制

TTL，time to live 的简称，即过期时间。

RabbitMQ可以对消息和队列两个维度来设置TTL。

任何消息中间件的容量和堆积都是有限的，如果有一些消息总是不被消费掉，那么需要有一种过期的机制来做兜底。

目前有两种方法可以设置消息的TTL。

  1. 通过Queue属性设置，队列中所有的消息使用相同的过期时间。
  2. 对消息自身进行单独的设置。每条消息的TTL可以不同。

如果两种方法一起使用，则消息的TTL以两者之间较小数值为准，通常来说，消息在队列中的生存时间一旦超过设置的TTL值时，就会变更“死信（Dead Message）”，消费者默认就无法再收到该消息。当然，“死信”也是可以被取出来消费的。

使用原生API操作

在队列上设置消息的过期时间

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

public class ProductTTL {
  public static void main(String[] args) throws Exception {

    ConnectionFactory factory = new ConnectionFactory();

    factory.setUri("amqp://root:123456@node1:5672/%2f");

    try (Connection connection = factory.newConnection();
        Channel channel = connection.createChannel(); ) {
      // 定义相关的参数
      Map<String, Object> param = new HashMap<>();

      // 设置队列的TTL，即此消息10秒后过期，
      param.put("x-message-ttl", 10000);
      // 设置队列的空闲时间，（如果队列没有消费者或者一直没有使用，队列可存活的时间）
      // 可以理解为没有消费者时，消息队列20秒后删除。
      param.put("x-expires", 20000);

      channel.queueDeclare("ttl.qu", false, false, false, param);

      for (int i = 0; i < 100; i++) {
        String sendMsg = "this is test msg :" + i;
        channel.basicPublish("", "ttl.qu", null, sendMsg.getBytes(StandardCharsets.UTF_8));
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
}

启动生产者

检查队列的信息

[root@nullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name   │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ttl.qu │ 100            │ 0                       │ 100      │ 0         │
└────────┴────────────────┴─────────────────────────┴──────────┴───────────┘

经过10秒后

[root@nullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name   │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ttl.qu │ 0              │ 0                       │ 0        │ 0         │
└────────┴────────────────┴─────────────────────────┴──────────┴───────────┘

再经过10秒后

[root@nullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...

经过操作后可以发现，此消息队列中的消息存活了10秒，然后消息就被清空了，由于队列的存活时间设置为20秒，没有消费者，所以经过了20秒后，队列就被清空了。

还可以将消息指定在消息中，效果相同

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

public class ProductTTLMsg {
  public static void main(String[] args) throws Exception {

    ConnectionFactory factory = new ConnectionFactory();
    factory.setUri("amqp://root:123456@node1:5672/%2f");
    try (Connection connection = factory.newConnection();
        Channel channel = connection.createChannel(); ) {
      // 定义相关的参数
      Map<String, Object> param = new HashMap<>();
      // 设置队列的空闲时间，（如果队列没有消费者或者一直没有使用，队列可存活的时间）
      // 可以理解为没有消费者时，消息队列20秒后删除。
      param.put("x-expires", 20000);
      channel.queueDeclare("ttl.qu", false, false, false, param);

      for (int i = 0; i < 100; i++) {
        String sendMsg = "this is test msg :" + i;
        // 在消息上指定存活时间为8秒
        AMQP.BasicProperties properties =
            new AMQP.BasicProperties().builder().expiration("8000").build();
        channel.basicPublish("", "ttl.qu", properties, sendMsg.getBytes(StandardCharsets.UTF_8));
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
}

还可以通过命令设置全局的TTL

rabbitmqctl set_policy TTL ".*" '{"message-ttl": 9000}' --apply-to queues

观察队列的policy信息

未设置策略前

[root@nullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers,policy  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┬────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │ policy │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┼────────┤
│ ttl.tmp.queue │ 0              │ 0                       │ 0        │ 0         │        │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┴────────┘

设置策略后

[root@nullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers,policy  --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┬────────┐
│ name          │ messages_ready │ messages_unacknowledged │ messages │ consumers │ policy │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┼────────┤
│ ttl.tmp.queue │ 0              │ 0                       │ 0        │ 0         │ TTL    │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┴────────┘

默认规则：

  1. 如果不设置TTL，则表示此消息不会过期；
  2. 如果TTL设置为0，则表示除非此时可以直接将消息投递到消费者，否则该消息会被立即丢弃。

http://www.kler.cn/a/463747.html

相关文章：

  • vue路由模式面试题
  • 学习路之VScode--自定义按键写注释（插件）
  • Appium 2.0：移动自动化测试的革新之旅
  • 【文献精读笔记】Explainability for Large Language Models: A Survey （大语言模型的可解释性综述）（三）
  • 迟来的前端面试经验
  • 逻辑推理算法
  • 缓存菜品的业务代码
  • 政策助力数字金融，CES Asia 2025展望科技新未来
  • AI 自动化编程对编程教育的影响
  • 每天五分钟机器学习：凸集
  • uni-app tab 双击事件监听
  • 前端小白学习之路-Vben探索 配置详情 - 2/50
  • 二、github基础
  • 【疑难杂症】 HarmonyOS NEXT中Axios库的响应拦截器无法拦截424状态码怎么办？
  • scala图书管理系统 【dao】软件包
  • A-Tune性能优化工具介绍
  • Python 语言实现 IP 地址转换
  • Linux Kernel Programming4
  • linux-24 文件管理(二）文件编辑，字符集，nano，文件删除，rm
  • 【Web安全】文件写入漏洞 ASP 网页病毒模拟（文件写入漏洞+FilesystemObject）
  • STM32单片机芯片与内部52 I2C 数据手册 寄存器
  • Prometheus 采集 JVM 数据
  • 从0到机器视觉工程师（二）:封装调用静态库和动态库
  • 重启ubuntu服务器，如何让springboot服务自动运行
  • 【SQL Server】教材数据库(1)
  • Javascript数据结构常见题目（一）