RabbitMQ - 延遲隊列

rabbitmq的延遲隊列，咱們能夠經過死信交換器來實現。
生產者發送消息，定義2秒後消息過時，消息就會進入死信交換器，最後到死信隊列。

// 定義隊列的名稱
public final static String QUEUE_NAME = "queue.scheduler";
// 定義交換器的名稱
public final static String EXCHANGE_NAME = "exchange.scheduler";
// 定義路由的名稱
public final static String ROUTE_NAME = "route.scheduler";
// 定義死信隊列的名稱
public final static String DLX_QUEUE_NAME = "scheduler.queue.name";
// 定義死信交換器的名稱
public final static String DLX_EXCHANGE_NAME = "scheduler.exchange.name";

public static void main(String[] args) throws IOException, TimeoutException {
    // 聲明一個鏈接工廠
    ConnectionFactory factory = new ConnectionFactory();
    // 建立一個與rabbitmq服務器的鏈接
    // 建立一個Channel
    try (Connection connection = factory.newConnection();
         Channel channel = connection.createChannel()) {
        // 定義交換器
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, false, false, null);
        Map<String, Object> arguments = new HashMap<String, Object>();
        arguments.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);
        arguments.put("x-message-ttl", 2000);
        // 定義隊列
        channel.queueDeclare(QUEUE_NAME, false, false, false, arguments);
        // 綁定隊列
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTE_NAME);
        // 定義死信交換器
        channel.exchangeDeclare(DLX_EXCHANGE_NAME, BuiltinExchangeType.DIRECT, false, false, null);
        // 定義死信隊列
        channel.queueDeclare(DLX_QUEUE_NAME, false, false, false, null);
        // 綁定死信隊列
        channel.queueBind(DLX_QUEUE_NAME, DLX_EXCHANGE_NAME, ROUTE_NAME);
        // 發送消息
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        channel.basicPublish(EXCHANGE_NAME, ROUTE_NAME, true, null, df.format(new Date()).getBytes());
    }
}

消費者，從私信隊列獲取消息，能夠獲得延遲後的消息。

public static void main(String[] args) throws IOException, TimeoutException {
    // 聲明一個鏈接工廠
    ConnectionFactory factory = new ConnectionFactory();
    // 建立一個與rabbitmq服務器的鏈接
    Connection connection = factory.newConnection();
    // 建立一個Channel
    Channel channel = connection.createChannel();
    System.out.println("Waiting for messages.");
    // 異步回調處理
    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
        String message = new String(delivery.getBody(), "UTF-8");
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println(df.format(new Date()) + " Received '" + message + "'");
    };
    // 接收消息
    channel.basicConsume(ProducerScheduler.DLX_QUEUE_NAME, true, deliverCallback, consumerTag -> {
    });
}

運行結果以下，達到了延遲隊列的效果。除此以外，還能夠用啓用延遲插件。