Spring Boot(十四)RabbitMQ延迟队列

一、前言

延迟队列的使用场景:1.未按时支付的订单,30分钟过期之后取消订单;2.给活跃度比较低的用户间隔N天之后推送消息,提高活跃度;3.过1分钟给新注册会员的用户,发送注册邮件等。

实现延迟队列的方式有两种:

  1. 通过消息过期后进入死信交换器,再由交换器转发到延迟消费队列,实现延迟功能;
  2. 使用rabbitmq-delayed-message-exchange插件实现延迟功能;

注意: 延迟插件rabbitmq-delayed-message-exchange是在RabbitMQ 3.5.7及以上的版本才支持的,依赖Erlang/OPT 18.0及以上运行环境。

由于使用死信交换器相对曲折,本文重点介绍第二种方式,使用rabbitmq-delayed-message-exchange插件完成延迟队列的功能。

二、安装延迟插件

1.1 下载插件

打开官网下载:http://www.rabbitmq.com/community-plugins.html

选择相应的对应的版本“3.7.x”点击下载。

注意: 下载的是.zip的安装包,下载完之后需要手动解压。

1.2 安装插件

拷贝插件到Docker:

docker cp D:\rabbitmq_delayed_message_exchange-20171201-3.7.x.ez rabbit:/plugins

RabbitMQ在Docker的安装,请参照本系列的上一篇文章:http://www.apigo.cn/2018/09/11/springboot13/

1.3 启动插件

进入docker内部:

docker exec -it rabbit /bin/bash

开启插件:

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

查询安装的所有插件:

rabbitmq-plugins list

安装正常,效果如下图:

重启RabbitMQ,使插件生效

docker restart rabbit

三、代码实现

3.1 配置队列

import com.example.rabbitmq.mq.DirectConfig;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class DelayedConfig {
    final static String QUEUE_NAME = "delayed.goods.order";
    final static String EXCHANGE_NAME = "delayedec";
    @Bean
    public Queue queue() {
        return new Queue(DelayedConfig.QUEUE_NAME);
    }

    // 配置默认的交换机
    @Bean
    CustomExchange customExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        //参数二为类型:必须是x-delayed-message
        return new CustomExchange(DelayedConfig.EXCHANGE_NAME, "x-delayed-message", true, false, args);
    }
    // 绑定队列到交换器
    @Bean
    Binding binding(Queue queue, CustomExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(DelayedConfig.QUEUE_NAME).noargs();
    }
}

3.2 发送消息

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.Date;

@Component
public class DelayedSender {
    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send(String msg) {
        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("发送时间:" + sf.format(new Date()));

        rabbitTemplate.convertAndSend(DelayedConfig.EXCHANGE_NAME, DelayedConfig.QUEUE_NAME, msg, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setHeader("x-delay", 3000);
                return message;
            }
        });
    }
}

3.3 消费消息

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.Date;

@Component
@RabbitListener(queues = "delayed.goods.order")
public class DelayedReceiver {
    @RabbitHandler
    public void process(String msg) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("接收时间:" + sdf.format(new Date()));
        System.out.println("消息内容:" + msg);
    }
}

3.4 测试队列

import com.example.rabbitmq.RabbitmqApplication;
import com.example.rabbitmq.mq.delayed.DelayedSender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.text.SimpleDateFormat;
import java.util.Date;

@RunWith(SpringRunner.class)
@SpringBootTest
public class DelayedTest {

    @Autowired
    private DelayedSender sender;

    @Test
    public void Test() throws InterruptedException {
        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd");
        sender.send("Hi Admin.");
        Thread.sleep(5 * 1000); //等待接收程序执行之后,再退出测试
    }
}

执行结果如下:

发送时间:2018-09-11 20:47:51
接收时间:2018-09-11 20:47:54
消息内容:Hi Admin.

完整代码访问我的GitHub:https://github.com/vipstone/springboot-example/tree/master/springboot-rabbitmq

四、总结

到此为止我们已经使用“rabbitmq-delayed-message-exchange”插件实现了延迟功能,但是需要注意的一点是,如果使用命令“rabbitmq-plugins disable rabbitmq_delayed_message_exchange”禁用了延迟插件,那么所有未发送的延迟消息都将丢失。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Java技术分享圈

JDBC、C3P0、DBCP、Druid 数据源连接池使用的对比总结.md

因为建立数据库连接是一个非常耗时、耗资源的行为,所以通过连接池预先同数据库建立一些连接,放在内存中,应用程序需要建立数据库连接时直接到连接池中申请一个就行,用完...

2103
来自专栏JetpropelledSnake

ELK学习笔记之Logstash和Filebeat解析对java异常堆栈下多行日志配置支持

logstash官方最新文档。 假设有几十台服务器,每台服务器要监控系统日志syslog、tomcat日志、nginx日志、mysql日志等等,监控OOM、内存...

3331
来自专栏SpringBoot 核心技术

第四十三章: 基于SpringBoot & RabbitMQ完成TopicExchange分布式消息消费

36215
来自专栏SpringBoot 核心技术

第三章:使用QueryDSL与SpringDataJPA完成Update&Delete

2892
来自专栏帘卷西风的专栏

使用Cmake生成跨平台项目编译解决方案

    项目最近有需求在windows下面运行,我花了几周时间将linux的服务器移植到windows下面,目前已经能够正常运行服务器,目前又有了新需求,两边的...

4232
来自专栏JetpropelledSnake

RESTful源码学习笔记之RPC和Restful深入理解

RPC 即远程过程调用(Remote Procedure Call Protocol,简称RPC),像调用本地服务(方法)一样调用服务器的服务(方法)。通常的实...

953
来自专栏互联网高可用架构

简单易用的消息队列框架的设计与实现

2322
来自专栏张戈的专栏

Linux系统crontab备份数据库执行不成功?可能是百分号%在作怪!

之前博客分享过一篇《Linux/vps 本地七天循环备份和七牛远程备份脚本》,我自己也一直在用。某天检查备份的时候,突然发现数据库的备份的压缩包是空的! 看了下...

3104
来自专栏葡萄城控件技术团队

Web API 持续集成:PostMan+Newman+Jenkins(图文讲解)

上篇文章我们已经完成了API测试工具选型,接下来是一系列周期性的开发测试过程:接口开发、检出代码、运行测试、记录结果、发送报告。为了快速发现问题,并减少重复过程...

1502
来自专栏Ryan Miao

使用Apache Server 的ab进行web请求压力测试

参考:http://www.cnblogs.com/spring3mvc/archive/2010/11/23/2414741.html 自己写代码经常是顺着逻...

3497

扫码关注云+社区

领取腾讯云代金券