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 -> {    });}

运行结果如下，达到了延迟队列的效果。除此之外，还可以用启用延迟插件。