RabbitMQ高級特性-TTL隊列/消息

TTL

time to live,即生存時間

RabbitMQ 支持消息的過時時間，能夠在發消息是指定

RabbitMQ 支持隊列的過時時間，從消息入隊開始計算，只要超過了隊列的超時時間配置，那麼消息會自動清除

代碼

package com.wyg.rabbitmq.javaclient.ttl;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * ttl隊列/消息
 * 
 * @author wyg0405@gmail.com
 * @date 2019-11-22 13:25
 * @since JDK1.8
 * @version V1.0
 */

public class Producer {
    private static final String HOST = "localhost";
    private static final int PORT = 5672;
    private static final String USERNAME = "guset";
    private static final String PASSWORD = "guset";

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST);
        factory.setVirtualHost("/");
        factory.setPort(PORT);
        factory.setUsername(USERNAME);
        factory.setPassword(PASSWORD);
        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();
        String exchangeName = "test_ttl_exchange";
        String routingKey = "ttl.abc";
        String queueName = "test_ttl_queue";

        // 申明exchange
        channel.exchangeDeclare(exchangeName, "topic");
        // 申明 queue
        channel.queueDeclare(queueName, true, false, false, null);

        Map<String, Object> arguments = new HashMap<>();
        // 隊列ttl，設置爲8s
        arguments.put("message-ttl", 8000);
        channel.queueBind(queueName, exchangeName, routingKey, arguments);

        for (int i = 0; i < 3; i++) {
            // expiration("10000") 設置消息8s過時,消息的ttl
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().deliveryMode(2).contentEncoding("UTF-8")
                .expiration("8000").build();

            String msg = "這是第" + i + "條ack消息";
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            channel.basicPublish(exchangeName, routingKey, false, props, msg.getBytes("UTF-8"));
        }

        channel.close();
        connection.close();
    }

}

運行結果(動圖)，觀察管理控制檯，到期自動清除