RabbitMQ 延时消息的实现（上）

我们在实际业务中有一些需要延时发送消息的场景，例如：

1、 家里有一台智能热水器，需要在30分钟后启动

2、 未付款的订单，15分钟后关闭

注意这里的场景是延时，不是定时。当然，解决了延时，定时就很简单了（定时=当前时刻+间隔时间）。

由于 rabbitmq 本身不支持延时队列（延时消息），所以要通过其他方式来实现。总的来说有三种：

1、 先存储到数据库，用定时任务扫描，登记时刻+延时时间，就是需要投递的时刻

2、 利用RabbitMQ的死信队列（Dead Letter Queue）实现

3、 利用rabbitmq-delayed-message-exchange插件

定时任务实现比较简单，此处略过。我们来看一下后两种方案分别怎么实现。

前提知识：我们可以在发送消息时指定单条消息的存活时间（Time To Live， TTL ）。也可以设置一个队列的消息过期时间。

这两种方式，当队列中的消息到达过期时间（比如30分钟）仍未被消费，就会被发送到队列的死信交换机（Dead Letter Exchange，DLX），被再次路由，此时再次路由到的队列就被称为死信队列（Dead Letter Queue）。需要注意，死信交换机和死信交换机都是基于其用途来描述的，它们实际上也是普通的交换机和普通的队列。如果队列没有指定DLX或者无法被路由到一个DLQ，则队列中过期的消息会被直接丢弃。

因此，我们可以利用消息TTL的特性，实现消息的延时投递。

1、设置单条消息的过期时间的方法：

AMQP.BasicProperties properties = new AMQP.BasicProperties. build er()

.deliveryMode(2) // 持久化 消息

.contentEncoding(“UTF-8”)

.expiration(“10000”) // TTL，10秒后没有被消费则被发送到DLX

.build();

channel .basicPublish(“”, “TEST_TTL_QUEUE”, properties, msg .getBytes()); //此处发送到 AMQP Default 这个默认的Direct类型的交换机，并路由到TEST_TTL_QUEUE队列

2、设置队列的消息过期时间的方法：

Map<String, Object> argss = new HashMap<String, Object>();

argss.put(“x-message-ttl”,6000); // TTL，6秒后没有被消费则被发送到DLX

channel.queueDeclare(“TEST_TTL_QUEUE”, false , false, false, argss);

注意：如果同时设置了消息的过期时间和队列的消息过期时间，则会取其中一个较小的值。比如消息设置5秒过期，队列设置消息10秒过期，则实际过期时间是5秒。

基于消息TTL，我们来看一下如何利用死信队列（DLQ）实现延时队列：

总体步骤：

1）创建一个交换机

2）创建一个队列，与上述交换机绑定，并且通过属性指定队列的死信交换机。

3）创建一个死信交换机

4）创建一个死信队列

4）将死信交换机绑定到死信队列

5）消费者监听死信队列

代码如下：

消费者：

因为此处使用默认的AMQP Default的Exchange，所以省略了第1）步，没有创建交换机。

这里用指定消息的TTL实现，所以设置队列TTL属性的代码注释了。

// 指定队列的死信交换机

Map<String,Object> arguments = new HashMap<String,Object>();

arguments.put( “x-dead-letter- exchange ” , “DLX_EXCHANGE” );

// arguments.put(“x-expires”,”9000″); // 设置队列的TTL

// 声明队列（默认交换机AMQP default，Direct）

channel.queueDeclare( “TEST_DLX_QUEUE” , false , false , false , arguments);

// 声明死信交换机

channel.exchangeDeclare( “DLX_EXCHANGE” , “topic” , false , false , false , null );

// 声明死信队列

channel.queueDeclare( “DLX_QUEUE” , false , false , false , null );

// 绑定，此处 Dead letter routing key 设置为 #，代表路由所有消息

channel.queueBind( “DLX_QUEUE” , “DLX_EXCHANGE” , “#” );

生产者：

String msg = “Hello world, Rabbit MQ, DLX MSG” ;

// 设置属性，消息10秒钟过期

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()

.deliveryMode(2) // 持久化消息

.contentEncoding( “UTF-8” )

.expiration( “10000” ) // TTL

.build();

// 发送消息

channel.basicPublish( “” , “TEST_DLX_QUEUE” , properties, msg.getBytes());

消息的流转流程

生产者——原交换机——原队列——（超过TTL之后）——死信交换机——死信队列——最终消费者

如图：

使用死信队列实现延时消息的缺点：

1） 如果统一用队列来设置消息的TTL，当梯度非常多的情况下，比如1分钟，2分钟，5分钟，10分钟，20分钟，30分钟……需要创建很多交换机和队列来路由消息。

2） 如果单独设置消息的TTL，则可能会造成队列中的消息阻塞——前一条消息没有出队（没有被消费），后面的消息无法投递。

3） 可能存在一定的时间误差。

下一篇文章，我们来讲一下如何使用rabbitmq的延时消息插件来实现延时消息。

RabbitMQ 延时消息的实现（下）

在RabbitMQ 3.5.7及以后的版本提供了一个插件（rabbitmq-delayed-message-exchange）来实现延时队列功能。同时插件依赖Erlang/OPT 18.0及以上。

插件源码地址：

插件下载地址：

1、进入插件目录

whereis rabbitmq

cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.12/plugins

2、下载插件

wget

如果下载的文件名带问号则需要改名，例如：

mv download_file?file_path=rabbitmq_delayed_message_exchange-0.0.1.ez rabbitmq_delayed_message_exchange-0.0.1.ez

3、启用插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

4、停用插件

rabbitmq-plugins disable rabbitmq_delayed_message_exchange

5、插件使用

通过声明一个x-delayed-message类型的exchange来使用delayed-messaging特性。x-delayed-message是插件提供的类型，并不是rabbitmq本身的（区别于direct、topic、fanout、headers）。

代码：

消费者（先启动）：

// 声明x-delayed-message类型的exchange

Map<String, Object> argss = new HashMap<String, Object>();

argss.put( “x-delayed-type” , “direct” );

channel.exchangeDeclare( “DELAY_EXCHANGE” , “x-delayed-message” , false ,

false , argss);

// 声明队列

channel.queueDeclare( “DELAY_QUEUE” , false , false , false , null );

// 绑定交换机与队列

channel.queueBind( “DELAY_QUEUE” , “DELAY_EXCHANGE” , “DELAY_KEY” );

// 创建消费者

Consumer consumer = new DefaultConsumer(channel) {

@Override

public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,

byte [] body) throws IOException {

String msg = new String(body, “UTF-8” );

SimpleDateFormat sf= new SimpleDateFormat( “yyyy-MM-dd HH:mm:ss.SSS” );

System. out .println( “收到消息：[” + msg + “]n接收时间：” +sf.format( new Date()));

}

};

// 开始获取消息

channel.basicConsume( “DELAY_QUEUE” , true , consumer);

生产者（后启动）：

// 延时投递，比如延时1分钟

Date now = new Date();

Calendar calendar = Calendar. getInstance ();

calendar.add(Calendar. MINUTE , +1); // 1分钟后投递

Date delayTime = calendar.getTime();

SimpleDateFormat sf = new SimpleDateFormat( “yyyy-MM-dd HH:mm:ss.SSS” );

String msg = “发送时间：” + sf.format(now) + “，投递时间：” + sf.format(delayTime);

// 延迟的间隔时间，目标时刻减去当前时刻

Map<String, Object> headers = new HashMap<String, Object>();

headers.put( “x-delay” , delayTime.getTime() – now.getTime());

AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder()

.headers(headers);

channel.basicPublish( “DELAY_EXCHANGE” , “DELAY_KEY” , props.build(),

msg.getBytes());

channel. close ();

conn.close();

控制台输出：

收到消息：[发送时间：2019-01-15 20:44:41.000，投递时间：2019-01-15 20:45:41.003]

接收时间：2019-01-15 20:45:41.064

2018已经过去过去，2019还想一成不变吗？拥抱变化，突破瓶颈，想要学习 Java 架构技术的朋友可以加我的群：725219329，群内每晚都会有阿里技术大牛讲解的最新Java架构技术。并会录制录播视频分享在群公告中，作为给广大朋友的加群的福利——分布式(Dubbo、Redis、RabbitMQ、Netty、RPC、Zookeeper、高并发、高可用架构)/微服务（Spring Boot、Spring Cloud）/源码（Spring、Mybatis）/性能优化（JVM、TomCat、MySQL）