Rabbitmq延迟队列学习

1概述

1.1延迟队列

延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费。

应用场景:

下单后，30分钟未支付，取消订单，回滚库存。

30分钟内字符,则从MQ消息队列中修改消息状态half-->可发送。

30分钟到, 未支付,则取消订单,回滚库存,删除消息。

在RabbitMQ中并未提供延迟队列功能。
但是可以使用：TTL+死信队列组合实现延迟队列的效果。

2代码示例

2.1生产端

package com.kikop.workpattern.topic.producer;

import com.kikop.common.utils.DateUtil;
import com.kikop.common.utils.RabbitMQUtils;
import com.kikop.config.RabbitMQConfig;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

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

/**
 * @author kikop
 * @version 1.0
 * @project Name: myrabbitmqdemo
 * @file Name: WeatherBureau
 * @desc 功能描述 消费消息
 * @date 2020/9/6
 * @time 18:46
 * @by IDE: IntelliJ IDEA
 */
public class DelayedQueueProducer {

    /**
     * 单独为某条消息设置超时
     *
     * @throws IOException
     * @throws TimeoutException
     */
    private static void createDelayQueueTest() throws IOException, TimeoutException {

        Connection connection = RabbitMQUtils.getConnection();
        final Channel channel = connection.createChannel();

        // 定义死信交换机
        boolean durable_dlx_exchange = true;
        channel.exchangeDeclare(RabbitMQConfig.EXCHANGE_DLX_TOPIC, BuiltinExchangeType.TOPIC, durable_dlx_exchange);


        // 1.过期队列及属性设置
        Map<String, Object> arguments = new HashMap<String, Object>();

        // 1.1.设置整个队列过期时间
        // arguments.put("x-message-ttl", 10000);

        // 1.2.设置队列的私信参数(队列过期的后处理,移到死信交换机中)
        arguments.put("x-dead-letter-exchange", RabbitMQConfig.EXCHANGE_DLX_TOPIC);
        arguments.put("x-dead-letter-routing-key", "rk_mydlx");

        // 1.3.创建持久化队列
        boolean durable_queue = true;
        channel.queueDeclare(RabbitMQConfig.QUEUE_DELAY_TOPIC_PRIVATE, durable_queue, false, false, arguments);

        // 2.创建持久化转发代理
        boolean durable_exchange = true;
        channel.exchangeDeclare(RabbitMQConfig.EXCHANGE_DELAY_TOPIC_PRIVATE, BuiltinExchangeType.TOPIC, durable_exchange);
        String routingKey = "mydelay";
        channel.queueBind(RabbitMQConfig.QUEUE_DELAY_TOPIC_PRIVATE, RabbitMQConfig.EXCHANGE_DELAY_TOPIC_PRIVATE, routingKey);

        // 3.设置消息持久化
        String message = "购票超时,订单编号:000001,锁定的库存编号为:000110,即将执行回滚,(增加库存,作废订单)操作!";
        int durable_msg = 2; // 设置消息是否持久化，1： 非持久化 2：持久化
        AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties().builder();
        properties.deliveryMode(durable_msg);  // 设置消息是否持久化，1： 非持久化 2：持久化

        // 4.发送 mock half消息
        properties.expiration(3 * 60 * 1000 + ""); // 过期时间3分钟
        System.out.println(DateUtil.getCurrentDateStr());
        channel.basicPublish(RabbitMQConfig.EXCHANGE_DELAY_TOPIC_PRIVATE, routingKey,
                properties.build(), message.getBytes("UTF-8"));

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


    public static void main(String[] args) throws Exception {

        createDelayQueueTest();
        return;
    }

}

2.2消费端

package com.kikop.workpattern.topic.consumer;

import com.kikop.common.utils.DateUtil;
import com.kikop.common.utils.RabbitMQUtils;
import com.kikop.config.RabbitMQConfig;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author kikop
 * @version 1.0
 * @project Name: myrabbitmqdemo
 * @file Name: BiaDu
 * @desc 功能描述 消费消息
 * @date 2020/9/6
 * @time 18:46
 * @by IDE: IntelliJ IDEA
 */
public class DelayedQueueConsumer {

    public static void main(String[] args) throws IOException, TimeoutException {

        delayConsumerTest();
        return;
    }


    private static void delayConsumerTest() throws IOException, TimeoutException {
        Connection connection = RabbitMQUtils.getConnection();
        final Channel channel = connection.createChannel();


        // begin_配置通道队列(可选,可在生产端完成)

        // 定义通道的非持久化队列
        channel.queueDeclare(RabbitMQConfig.QUEUE_DLX_TOPIC, false, false, false, null);

        // 定义需要消费的私信队列 QUEUE_DLX_TOPIC,从关联的死信交换机中抓取
        // 增加现有的一个转发代理,代理 EXCHANGE_DLX_TOPIC 必须先存着
        //  queueBind用于将队列与交换机绑定
        // 参数1：队列名
        // 参数2：交互机名
        // 参数三：路由key
        channel.queueBind(RabbitMQConfig.QUEUE_DLX_TOPIC, RabbitMQConfig.EXCHANGE_DLX_TOPIC, "rk_mydlx");
        // end_配置通道队列(可选)

        channel.basicQos(1);

        // 只消费死信队列里面的消息
        channel.basicConsume(RabbitMQConfig.QUEUE_DLX_TOPIC, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(DateUtil.getCurrentDateStr() + "@信息：" + new String(body));

                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }

}

2.3测试

// 9.延迟队列(TTL+DLX)

// 实际上是私有的
public static final String EXCHANGE_DELAY_TOPIC_PRIVATE = "ex_delay_topic_private";
public static final String QUEUE_DELAY_TOPIC_PRIVATE = "queue_delay_topic_private";


public static final String EXCHANGE_DLX_TOPIC = "ex_dlx_topic";
public static final String QUEUE_DLX_TOPIC = "queue_dlx_topic";

生产端:2021-03-28 23:02:47

消费端:2021-03-28 23:05:47@信息：购票超时,订单编号:000001,锁定的库存编号为:000110,即将执行回滚,(增加库存,作废订单)操作!

参考

1RabbitMQ消息最终一致性解决方案

https://blog.csdn.net/weixin_43466542/article/details/101676944

2.RabbitMq(十一) 死信交换机DLX介绍及使用

https://blog.csdn.net/liuhenghui5201/article/details/107538775

2021-03-28_Rabbitmq延迟队列学习