RabbitMQ代碼篇之過期時間，死信隊列，延遲隊列，優(yōu)先級隊列的基本使用

作者：Java4ye 2024-03-18 00:00:03

開發(fā) 前端

創(chuàng)建隊列時設置，消息會被排序加入到隊列頭部，短的在前，發(fā)送消息時設置，時間到期不會立刻刪除，而是在推送消息時刪除。

這篇是代碼篇，略過 Direct，F(xiàn)anout，Topic 這幾種基本隊列的使用介紹，但是 Github 倉庫上可以查看到的。

圖片

<(￣︶￣)↗[GO!]

沖沖沖~

死信隊列

死信是指有這三個特點的消息

消息被拒絕，且沒有重新入隊（投遞）
消息過期
消息隊列滿了

// nack 返回 false，并放棄重新回到隊列
channel.basicNack(deliveryTag, false, false);

//拒絕，不重新入隊列
channel.basicReject(deliveryTag, false);

死信交換機 —— DLX：Dead-Letter-Exchange

@Bean
public DirectExchange directExchange2() {
    /**
     * 交換機名，后面兩個是默認值就：持久化，不自動刪除
     */
    return new DirectExchange(RabbitMQConstants.DIRECT_EXCHANGE2, true, false);
}

@Bean
public Queue directQueue2() {
    return QueueBuilder
            .durable(RabbitMQConstants.DIRECT_QUEUE2)
            .deadLetterExchange(RabbitMQConstants.DLX_EXCHANGE)
            .deadLetterRoutingKey(RabbitMQConstants.DLX_ROUTING_KEY)
            .build();
}

/**
 * 將隊列綁定到交換機上
 *
 * @return
 */
@Bean
public Binding directBinding2() {
    return BindingBuilder.
            bind(directQueue2()).
            to(directExchange2()).
            with(RabbitMQConstants.DIRECT_BINDING_KEY2);
}

過期時間

兩種設置方式

創(chuàng)建隊列時設置，消息會被排序加入到隊列頭部，短的在前
發(fā)送消息時設置，時間到期不會立刻刪除，而是在推送消息時刪除

同時設置的話，過期時間已短的為準

/**
 * 創(chuàng)建隊列時設置
 * @return
 */
@Bean
public Queue ttlQueue() {
    Map<String, Object> args = new HashMap<>();
    //設置消息過期時間
    args.put("x-message-ttl", 5000);
    //設置死信交換機
    args.put("x-dead-letter-exchange", RabbitMQConstants.DLX_EXCHANGE);
    //設置死信 routing_key
    args.put("x-dead-letter-routing-key", RabbitMQConstants.DLX_ROUTING_KEY);
    return new Queue(RabbitMQConstants.TTL_QUEUE, true, false, false, args);
}

/**
 * 發(fā)送消息時設置
 * @return
 */
public void sendMessage2() throws JsonProcessingException {
    User user = new User();
    ObjectMapper objectMapper = new ObjectMapper();
    byte[] bytes = objectMapper.writeValueAsBytes(user);

    // 10 s 后過期
    Message message =
            MessageBuilder.withBody(bytes)
                    .setExpiration("10000").build();


    // 交換機，路由鍵，信息
    rabbitTemplate.convertAndSend(
            RabbitMQConstants.DIRECT_EXCHANGE2,
            RabbitMQConstants.DIRECT_ROUTING_KEY2,
            message
    );

}

效果演示

圖片

隊列 5 秒延遲的效果

圖片

延遲隊列

兩種方案實現(xiàn)

利用死信隊列+過期時間去處理，消息過期被轉發(fā)到死信交換機，死信交換機路由到死信隊列進行處理
使用插件 rabbitmq_delayed_message_exchange

插件可以在這里找到 ??

https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

圖片

其他插件：https://www.rabbitmq.com/community-plugins.html

圖片

安裝插件步驟

拷貝文件

docker cp . rabbitmq:/plugins

進入容器

docker exec -it rabbitmq /bin/bash

開啟插件支持

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

// 延遲交換機
    @Bean
    public CustomExchange delayedExchange() {
        Map<String, Object> args = new HashMap<>();
        // 交換機類型
        args.put("x-delayed-type", "direct");
        return new CustomExchange(RabbitMQConstants.DELAY_EXCHANGE, "x-delayed-message", true, false, args);
    }

    public void sendMessage() throws JsonProcessingException {
        User user = new User();
        ObjectMapper objectMapper = new ObjectMapper();
        byte[] bytes = objectMapper.writeValueAsBytes(user);

        //延遲 10 s
        Message message = MessageBuilder.withBody(bytes).setHeader("x-delay", 10_000).build();

        // 交換機，路由鍵，信息
        rabbitTemplate.convertAndSend(
                RabbitMQConstants.DELAY_EXCHANGE,
                RabbitMQConstants.DELAY_ROUTING_KEY,
                message
        );
    }

效果演示

圖片

10 秒的延遲

圖片

優(yōu)先級隊列

x-max-priority 可以設置在隊列和消息上

設置在隊列上，表示該隊列是個優(yōu)先級隊列，同時，消息的最大優(yōu)先級無法超過隊列設置的上限
發(fā)送消息時，帶上優(yōu)先級，在消息堆積的情況下，優(yōu)先級高的會先被消費；

@Bean
public Queue directPriorityQueue() {
    Map<String, Object> args = new HashMap<>();
   // 最大優(yōu)先級
    args.put("x-max-priority", 10);
    return new Queue(RabbitMQConstants.PRIORITY_QUEUE, true, false, false, args);
}

public void sendMessage() throws Exception {
        User user = new User();
        ObjectMapper objectMapper = new ObjectMapper();

        // 交換機，路由鍵，信息, 優(yōu)先級 5
        retryRabbitTemplate.convertAndSend(
                RabbitMQConstants.PRIORITY_EXCHANGE,
                RabbitMQConstants.PRIORITY_ROUTING_KEY,
                MessageBuilder.withBody( objectMapper.writeValueAsBytes(user.setPriority(5))).setPriority(5).build()
        );

    }