RabbitMQ消息追踪、TTL、死信队列、延迟队列

2024-01-24 08:24:10

消息追踪

消息跟踪，排查问题。追踪消息的生产和消费

Firehose

Firehose 的原理是将生产者投递给RabbitMQ 的消息，或者RabbitMQ 投递给消费者的消息按照指定的格式发送到默认的交换器上。这个默认的交换器的名称为 amq.rabbitmq.trace ，它是一个 topic 类型的交换器。发送到这个交换器上的消息的路由键为 publish.{exchangename} 和 deliver. {queuename} 。其中 exchangename 和 queuename 为交换器和队列的名称，分别对应生产者投递到交

换器的消息和消费者从队列中获取的消息。

默认是非持久化的，且对性能有影响，生产环境不用，默认也是关闭的

rabbitmq_tracing

可视化

TTL机制

使用场景：
在京东下单，订单创建成功，等待支付，一般会给30分钟的时间，开始倒计时。如果在这段时间内用户没有支付，则默认订单取消。

该如何实现？

定期轮询（数据库等）

用户下单成功，将订单信息放入数据库，同时将支付状态放入数据库，用户付款更改数据库状态。定期轮询数据库支付状态，如果超过30分钟就将该订单取消。优点：设计实现简单缺点：需要对数据库进行大量的IO操作，效率低下。

Timer

缺点：

Timers没有持久化机制.
Timers不灵活 (只可以设置开始时间和重复间隔，对等待支付貌似够用)
Timers 不能利用线程池，一个timer一个线程
Timers没有真正的管理计划

ScheduledExecutorService

优点：可以多线程执行，一定程度上避免任务间互相影响，单个任务异常不影响其它任务。
在高并发的情况下，不建议使用定时任务去做，因为太浪费服务器性能，不建议。

RabbitMQ

使用TTL

Quartz

Redis Zset+过期时间

JCronTab

SchedulerX

TTL介绍

TTL，Time to Live 的简称，即过期时间。 RabbitMQ 可以对消息和队列两个维度来设置TTL。

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

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

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

如果两种方法一起使用，则消息的TTL 以两者之间较小数值为准。

通常来讲，消息在队列中的生存时间一旦超过设置的TTL 值时，就会变成“死信”(Dead Message)，消费者默认就无法再收到该消息。当然，“死信”也是可以被取出来消费的，下一小节我们会讲解。

RabbitMQ TTL实战

通过原生API设置

public class Producer {
    public static void main(String[] args) throws NoSuchAlgorithmException, KeyManagementException, URISyntaxException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUri("amqp://root:123456@10.211.55.11:5672/%2f");

        try (final Connection connection = factory.newConnection();
             final Channel channel = connection.createChannel()) {

            Map<String, Object> arguments = new HashMap<>();
//            消息队列中消息过期时间，30s
            arguments.put("x-message-ttl", 10 * 1000);
//            如果消息队列没有消费者，则10s后消息过期，消息队列也删除
//            arguments.put("x-expires", 10 * 1000);
            arguments.put("x-expires", 60 * 1000);

            channel.queueDeclare("queue.ttl.waiting",
                    true,
                    false,
                    false,
                    arguments);

            channel.exchangeDeclare("ex.ttl.waiting",
                    "direct",
                    true,
                    false,
                    null);

            channel.queueBind("queue.ttl.waiting", "ex.ttl.waiting", "key.ttl.waiting");

            final AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .contentEncoding("utf-8")
                    .deliveryMode(2)   // 持久化的消息
                    .build();

            channel.basicPublish("ex.ttl.waiting",
                    "key.ttl.waiting",
                    null,
                    "等待的订单号".getBytes("utf-8"));
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}

关键参数：

x-message-ttl
x-expires（消息队列很重要，就不要设置，否则如果没有消息，达到过期时间会删掉队列）

通过命令行方式设置全局TTL，执行如下命令：

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

设置队列中的消息过期时间30秒

通过restful api方式设置

默认规则：

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

注意理解 message-ttl 、 x-expires 这两个参数的区别，有不同的含义。但是这两个参数属性都遵循上面的默认规则。一般TTL相关的参数单位都是毫秒（ms）

SpringBoot方式设置

todo

死信队列

TTL可以用于订单支付的时候，如果30分钟没有支付，~~消息就删掉~~
事实上，一会直接将消息删除，而是把消息加入死信队列，发送给其他外卖小哥。

用户下单，调用订单服务，然后订单服务调用派单系统通知外卖人员送单，这时候订单系统与派单系统采用 MQ异步通讯。

在定义业务队列时可以考虑指定一个死信交换机，并绑定一个死信队列。当消息变成死信时，该消息就会被发送到该死信队列上，这样方便我们查看消息失败的原因。

DLX，全称为Dead-Letter-Exchange，死信交换器。消息在一个队列中变成死信（Dead Letter）之后，被重新发送到一个特殊的交换器（DLX）中，同时，绑定DLX的队列就称为“死信队列”。

以下几种情况导致消息变为死信：

消息被拒绝（Basic.Reject/Basic.Nack），并且设置requeue参数为false；
消息过期；
队列达到最大长度。

对于RabbitMQ 来说，DLX 是一个非常有用的特性。

它可以处理异常情况下，消息不能够被费者正确消费(消费者调用了Basic.Nack 或者Basic.Reject)而被置入死信队列中的情况，后续分析程序可以通过消费这个死信队列中的内容来分析当时所遇到的异常情况，进而可以改善和优化系统。

RabbitMq死信队列实战

public class Producer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUri("amqp://root:123456@10.211.55.11:5672/%2f");

        try (final Connection connection = factory.newConnection();
             final Channel channel = connection.createChannel()) {

//            正常业务的交换器
            channel.exchangeDeclare("ex.biz", "direct", true);
//            声明死信交换器 DLX
            channel.exchangeDeclare("ex.dlx", "direct", true);
//            声明队列做死信队列
            channel.queueDeclare("queue.dlx", true, false, false, null);
//            绑定死信交换器和死信队列
            channel.queueBind("queue.dlx", "ex.dlx", "key.dlx");

            Map<String, Object> arguments = new HashMap<>();
//            指定消息队列中的消息过期时间
            arguments.put("x-message-ttl", 10000);

//            指定过期消息通过死信交换器发送到死信队列，死信交换器的名称，DLX
            arguments.put("x-dead-letter-exchange", "ex.dlx");
//            指定死信交换器的路由键
            arguments.put("x-dead-letter-routing-key", "key.dlx");

            channel.queueDeclare("queue.biz", true, false, false, arguments);
//            绑定业务的交换器和消息队列
            channel.queueBind("queue.biz", "ex.biz", "key.biz");

            channel.basicPublish("ex.biz", "key.biz", null, "orderid.45789987678".getBytes());

        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}

死信队列基于TTL，过期后将消息发送到死信队列
死信队列本身是一个普通队列，只是在声明业务队列时，设置私信队列参数，绑定死信需要发送到哪个队列。

SpringBoot方式死信队列实战

todo

延迟队列

延迟消息是指的消息发送出去后并不想立即就被消费，而是需要等（指定的）一段时间后才触发消费。例如下面的业务场景：在支付宝上面买电影票，锁定了一个座位后系统默认会帮你保留15分钟时间，如果15分钟后还没付款那么不好意思系统会自动把座位释放掉。怎么实现类似的功能呢？

可以用定时任务每分钟扫一次，发现有占座超过15分钟还没付款的就释放掉。但是这样做很低效，很多时候做的都是些无用功；
可以用分布式锁、分布式缓存的被动过期时间，15分钟过期后锁也释放了，缓存key也不存在了；
还可以用延迟队列，锁座成功后会发送1条延迟消息，这条消息15分钟后才会被消费，消费的过程就是检查这个座位是否已经是“已付款”状态;

你在公司的协同办公系统上面预约了一个会议，邀请汪产品和陈序员今晚22点准时参加会有。系统还比较智能，除了默认发会议邀请的邮件告知参会者以外，到了今晚21:45分的时候（提前15分钟）就会通知提醒参会人员做好参会准备，会议马上开始...

同样的，这也可以通过轮询“会议预定表”来实现，比如我每分钟跑一次定时任务看看当前有哪些会议即将开始了。当然也可以通过延迟消息来实现，预定会议以后系统投递一条延迟消息，而这条消息比较特殊不会立马被消费，而是延迟到指定时间后再触发消费动作（发通知提醒参会人准备）。不过遗憾的是，在AMQP协议和RabbitMQ中都没有相关的规定和实现。不过，我们似乎可以借助上一小节介绍的“死信队列”来变相的实现。

如何通过死信队列变相实现上面的延迟消息功能？

可以给要延迟的消息一个TTL，到期后，发送到死信队列，客户端从死信队列消费消息。
但是这样实现有个问题，由于消息队列是个队列，只支持FIFO。如果队列里有4个消息顺序如下，右边的先出来：
40sTTL，10sTTL，20sTTL，30sTTL
那么30sTTL的消息出来后，才能检查20sTTL的消息，问题是20sTTL的消息早就过期了。
即必须保证业务上的过期时间是一致的，如果想上面那种业务过期时间不一致，就不能用死信队列。
或者把相同过期时间的放入同一个死信队列

可以使用rabbitmq_delayed_message_exchange插件实现。

这里和TTL方式有个很大的不同就是TTL存放消息在死信队列(delayqueue)里，二基于插件存放消息在延时交换机里(x-delayed-message exchange)。

生产者将消息(msg)和路由键(routekey)发送指定的延时交换机(exchange)上
延时交换机(exchange)存储消息等待消息到期根据路由键(routekey)找到绑定自己的队列 (queue)并把消息给它
队列(queue)再把消息发送给监听它的消费者(customer）

消息缓存到延迟交换器中

下载插件下载地址：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases
安装插件将插件拷贝到rabbitmq-server的安装路径：/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.4/plugins
启用插件

rabbitmq-plugins list 
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

重启rabbitmq-server

systemctl restart rabbitmq-server

编写代码，首先是SpringBootApplication主入口类
application.properties文件

spring.application.name=delayed_exchange
spring.rabbitmq.host=node1
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=root
spring.rabbitmq.password=123456
spring.rabbitmq.port=5672
# 设置手动确认消息
#spring.rabbitmq.listener.simple.acknowledge-mode=manual

@SpringBootApplication
public class Demo19DelayedExchangeApplication {

    public static void main(String[] args) {
        SpringApplication.run(Demo19DelayedExchangeApplication.class, args);
    }
}

RabbitMQ的对象配置

@Configuration
@EnableRabbit
@ComponentScan("com.lagou.rabbitmq.demo")
public class RabbitConfig {

    @Bean
    public Queue queue() {
        return new Queue("queue.delayed", true, false, false, null);
    }

    @Bean
    public Exchange exchange() {

        Map<String, Object> arguments = new HashMap<>();
        // 使用x-delayed-type指定交换器的类型
        arguments.put("x-delayed-type", ExchangeTypes.DIRECT);
        // 使用x-delayed-message表示使用delayed exchange插件处理消息
        return new CustomExchange("ex.delayed", "x-delayed-message", true, false, arguments);
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange()).with("key.delayed").noargs();
    }

    @Bean
    @Autowired
    public RabbitAdmin rabbitAdmin(ConnectionFactory factory) {
        return new RabbitAdmin(factory);
    }

    @Bean
    @Autowired
    public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
        return new RabbitTemplate(factory);
    }

    @Bean
    @Autowired
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory
                = new SimpleRabbitListenerContainerFactory();

        factory.setConnectionFactory(connectionFactory);
        //
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        return factory;
    }

}

使用推消息模式接收延迟队列的广播

@Component
public class MyMeetingListener {

    @RabbitListener(queues = "queue.delayed")
    public void onMessage(Message message, Channel channel) throws IOException {

        System.out.println(new String(message.getBody(), message.getMessageProperties().getContentEncoding()));

        // 消息确认
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

开发RestController，用于向延迟队列发送消息，并指定延迟的时长

@RestController
public class DelayedController {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @RequestMapping("/meeting/{second}")
    public String bookMeeting(@PathVariable Integer second) throws UnsupportedEncodingException {

        final MessageProperties messageProperties = MessagePropertiesBuilder.newInstance()
                // 设置消息的过期时间
                .setHeader("x-delay", (second - 10) * 1000)
                .setContentEncoding("utf-8")
                .build();

        final Message message = MessageBuilder
                .withBody("还有10s开始开会了".getBytes("utf-8"))
                .andProperties(messageProperties)
                .build();

        amqpTemplate.send("ex.delayed", "key.delayed", message);

        return "会议定好了";
    }
}

pom.xml添加依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.8.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.lagou.rabbitmq.demo</groupId>
    <artifactId>demo_19_delayed_exchange</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>demo_19_delayed_exchange</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>11</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

结果：按照时长倒序发送请求，结果时间先到的先消费。

码农公寓