Spring Boot(十四)RabbitMQ延迟队列
一、前言
延迟队列的使用场景:1.未按时支付的订单,30分钟过期之后取消订单;2.给活跃度比较低的用户间隔N天之后推送消息,提高活跃度;3.过1分钟给新注册会员的用户,发送注册邮件等。
实现延迟队列的方式有两种:
- 通过消息过期后进入死信交换器,再由交换器转发到延迟消费队列,实现延迟功能;
- 使用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";@Beanpublic Queue queue() {return new Queue(DelayedConfig.QUEUE_NAME);}// 配置默认的交换机@BeanCustomExchange customExchange() {Map<String, Object> args = new HashMap<>();args.put("x-delayed-type", "direct");//参数二为类型:必须是x-delayed-messagereturn new CustomExchange(DelayedConfig.EXCHANGE_NAME, "x-delayed-message", true, false, args);}// 绑定队列到交换器@BeanBinding 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 {@Autowiredprivate 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() {@Overridepublic 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 {@RabbitHandlerpublic 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 {@Autowiredprivate DelayedSender sender;@Testpublic 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”禁用了延迟插件,那么所有未发送的延迟消息都将丢失。
转载于:https://www.cnblogs.com/vipstone/p/9967649.html
Spring Boot(十四)RabbitMQ延迟队列相关推荐
- Spring Boot(十四):spring boot整合shiro-登录认证和权限管理
Spring Boot(十四):spring boot整合shiro-登录认证和权限管理 使用Spring Boot集成Apache Shiro.安全应该是互联网公司的一道生命线,几乎任何的公司都会涉 ...
- (转)Spring Boot (十四): Spring Boot 整合 Shiro-登录认证和权限管理
http://www.ityouknow.com/springboot/2017/06/26/spring-boot-shiro.html 这篇文章我们来学习如何使用 Spring Boot 集成 A ...
- rabbitmq 延迟队列_Delayed Message 插件实现 RabbitMQ 延迟队列
延迟队列是为了存放那些延迟执行的消息,待消息过期之后消费端从队列里拿出来执行. DLX + TTL 方式存在的时序问题 对于延迟队列不管是 AMQP 协议或者 RabbitMQ 本身是不支持的,之前有 ...
- Delayed Message 插件实现 RabbitMQ 延迟队列
延迟队列是为了存放那些延迟执行的消息,待消息过期之后消费端从队列里拿出来执行. DLX + TTL 方式存在的时序问题 对于延迟队列不管是 AMQP 协议或者 RabbitMQ 本身是不支持的,之前有 ...
- RabbitMQ 延迟队列-对于入门来说可以快速上手
RabbitMQ 延迟队列-非常非常实用 RabbitMQ 延迟队列-非常非常实用 一.使用场景 二.消息延迟推送的实现 三.项目具体实现 RabbitMQ 延迟队列-非常非常实用 一.使用场景 ...
- 谷粒商城笔记+踩坑(22)——库存自动解锁。RabbitMQ延迟队列
导航: 谷粒商城笔记+踩坑汇总篇 目录 1 业务流程,订单失败后自动回滚解锁库存 可靠消息+最终一致性方案 2[仓库服务]RabbitMQ环境准备 2.1 导入依赖 2.2 yml配置RabbitMQ ...
- Spring Boot中使用RabbitMQ
很久没有写Spring Boot的内容了,正好最近在写Spring Cloud Bus的内容,因为内容会有一些相关性,所以先补一篇关于AMQP的整合. Message Broker与AMQP简介 Me ...
- (转)Spring Boot (十):邮件服务
http://www.ityouknow.com/springboot/2017/05/06/spring-boot-mail.html Spring Boot 仍然在狂速发展,才几个多月没有关注,现 ...
- Spring Boot (十五): Spring Boot + Jpa + Thymeleaf 增删改查示例
<p>这篇文章介绍如何使用 Jpa 和 Thymeleaf 做一个增删改查的示例.</p> 先和大家聊聊我为什么喜欢写这种脚手架的项目,在我学习一门新技术的时候,总是想快速的搭 ...
- RabbitMQ 延迟队列实现定时任务的正确姿势,你学会了么?
以下文章来源方志朋的博客,回复"666"获面试宝典 场景 开发中经常需要用到定时任务,对于商城来说,定时任务尤其多,比如优惠券定时过期.订单定时关闭.微信支付2小时未支付关闭订单等 ...
最新文章
- layer-list:Android中layer-list使用详解
- 计算机动画火柴人作业,(Flash期末作品综合实验报告.doc
- python写一个游戏多少代码-使用50行Python代码从零开始实现一个AI平衡小游戏
- CSP认证201703-1	分蛋糕[C++题解]:遍历
- 玩转控件:重写/重绘Dev中MessageBox弹窗控件
- linux配置本地yum(CentOS7)
- textarea中插入标签_HTMLCSS学习笔记(二)-- HTML表单标签
- 用遗传算法求3维函数 的最小值_遗传算法可视化项目(4):遗传算法
- 计算机网络原理html,计算机网络原理与应用html..ppt
- python xgboost_告别无聊循环!Python帮你自动化处理文件
- java时间管理番茄时钟小程序源码
- keil5的基本使用
- linux驱动开发期末测试,超星尔雅Linux驱动开发实训期末测试答案
- Spring(SpringBoot)--事务传播机制--应用/实例/详解
- 如何将txt格式转epub格式
- #bzoj1526#分梨子(乱搞)
- GScoolink GSV2201 TypeC/DP to HDMI2.0
- AR捡取纹理功能实现 colARmix涂涂乐
- CF1578F Framing Pictures 旋转卡壳+积分
- 小白怎么学习python