草庐IT

RabbitMQ实现延迟发送消息

gt_xz 2023-03-28 原文

前言

最近在做一个可以根据用户选择的时间,实现微信推送订阅消息的功能,突然想到rabbitmq好像可以实现这个功能,本着试试的心态开始研究,第一个想到的就是使用死信队列

死信队列

何为死信队列,其实rabbitmq本身并不能实现延迟发送消息的功能,不过因为本身有着队列ttl+死信exchange的机制,可以借助这个机制实现延迟发送消息

原理就是,用户发送消息到一个队列上并设置过期时间,但是这个队列没有消费者,到了过期时间,就由该队列绑定的死信exchange根据路由key的方式发送到另一个队列上,并消费,从而实现延迟发送

注:消息不仅只有超时才会成为死信,还有可能消息被消费者reject或者返回nack

理论上很简单,开始实践

导入依赖

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

创建配置类

1.声明死信队列,死信交换机,并绑定

2.声明普通队列,普通交换机,并绑定

3.普通队列需设置死信交换机,以及死信路由key

前两步很简单,就不贴代码了,需要注意第三步

Map<String, Object> args = new HashMap<>();
//设置死信交换机
args.put("x-dead-letter-exchange", "死信交换机名称");
//设置死信 routing_key
args.put("x-dead-letter-routing-key", "死信路由key");
return new Queue("普通队列", true, false, false, args);
创建之后,就需要一个生产者,发送消息

String id = UUID.randomUUID().toString();
logger.info("传递id消息========",id);
rabbitTemplate.convertAndSend(
"普通交换机名称","绑定普通队列路由key",id,
message -> {
//设置过期时间(毫秒)
message.getMessageProperties().setExpiration("过期时间");
return message;
});
生产消息后,创建消费者配置类

@RabbitListener(queues = "死信队列名称")
public void receiveMsg(Message message) {
String msg = new String(message.getBody());
logger.info("接收到消息:{}", msg);
}
至此死信队列完成,测试一下,果然能实现延迟发送,不出意外的话马上要出意外了,我的最初需求是可以根据客户选择的时间推送消息,死信队列不就只能固定时间吗,我要是有别的时间,还能重新创建一个队列发送消息,不然可能会出现一种情况,比如第一个用户发送了一个10秒的消息,第二个用户发送一个5秒的消息,那么在第一条消息成为死信之前,后面的消息即使过期也不会投递为死信,所以只能一个交换机绑定的普通队列对应一个时间,但是又不想创建那么多,非常麻烦不说,关键是解决不了我的需求;

于是在网上找资料,看到rabbitmq推出的一个插件 rabbitmq-delayed-message-exchange,这时候才明白,死信队列设计的初衷只是为了存储那些没有被正常消费的消息,便于重新发送,不至于出现消息丢失等情况,而 rabbitmq-delayed-message-exchange是专门用于发送延迟消息的,于是开始研究插件

 插件:rabbitmq-delayed-message-exchange

安装

要想使用插件肯定是先下载啦,网上教程有很多,我用的比较方便简单的方式下载,因为我是linux系统,所以直接使用命令下载

wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/3.8.9/rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez
注意:需要安装自己rabbitmq对应版本的插件

安装完毕启动

rabbitmq-plugins enable rabbitmq_delayed_message_exchange
这时候执行命令就可以看到有这个插件了

rabbitmq-plugins list


能看到这个插件,恭喜你安装成功,真聪明,接着往下走

代码配置

1.首先需声明队列,交换机根据路由key绑定

2.使用延迟消息交换器需要声明一个 x-delayed-message 类型的交换器

类似这样

Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
//自定义交换机
return new CustomExchange("延迟消息交换机", "x-delayed-message", false, false, args);
然后就可以定义生产者发送消息了

rabbitTemplate.convertAndSend("延迟消息交换机", "路由Key", "消息", message -> {
// 设置过期时间
message.getMessageProperties().setDelay("过期时间");
return message;
});
消费者同上

实验一下发现,不同时间也可以正常消费,这是因为这个插件可以让消息延迟性绑定到消息本身上,使的每个消息有自己的过期时间,原以为问题解决了,没想到又出问题了,他这个时间并不是随意的(网上说不能超过1个月),是有一个限制时间,而且性能要比原生的要差一点,不能忍,于是又想了个办法,定时任务

定时任务

定时任务的原理就很简单了,因为我的发送时间是存在数据库的,所以每过一段时间查询一下是否有需要发送的消息,有则发送,并消费

实现很简单

启动类添加注解

@EnableScheduling
还是老样子,声明队列,绑定交换机

定时任务的语法可以在网上找,需要用到@Scheduled注解,放在需要定时的方法上,可以实现定时执行

总结

至此问题基本解决了,不过可能会造成比用户指定的时间晚几秒推送的情况,不过我的功能并不需要这么精准,所以使用这种方法无非是最好的方式,三种方法都有使用的场景,没有好与不好,根据自己的业务选择就好

有关RabbitMQ实现延迟发送消息的更多相关文章

  1. ruby - 如何根据特征实现 FactoryGirl 的条件行为 - 2

    我有一个用户工厂。我希望默认情况下确认用户。但是鉴于unconfirmed特征,我不希望它们被确认。虽然我有一个基于实现细节而不是抽象的工作实现,但我想知道如何正确地做到这一点。factory:userdoafter(:create)do|user,evaluator|#unwantedimplementationdetailshereunlessFactoryGirl.factories[:user].defined_traits.map(&:name).include?(:unconfirmed)user.confirm!endendtrait:unconfirmeddoenden

  2. ruby-on-rails - 如何在 Rails View 上显示错误消息? - 2

    我是rails的新手,想在form字段上应用验证。myviewsnew.html.erb.....模拟.rbclassSimulation{:in=>1..25,:message=>'Therowmustbebetween1and25'}end模拟Controller.rbclassSimulationsController我想检查模型类中row字段的整数范围,如果不在范围内则返回错误信息。我可以检查上面代码的范围,但无法返回错误消息提前致谢 最佳答案 关键是您使用的是模型表单,一种显示ActiveRecord模型实例属性的表单。c

  3. jquery - 我的 jquery AJAX POST 请求无需发送 Authenticity Token (Rails) - 2

    rails中是否有任何规定允许站点的所有AJAXPOST请求在没有authenticity_token的情况下通过?我有一个调用Controller方法的JqueryPOSTajax调用,但我没有在其中放置任何真实性代码,但调用成功。我的ApplicationController确实有'request_forgery_protection'并且我已经改变了config.action_controller.consider_all_requests_local在我的environments/development.rb中为false我还搜索了我的代码以确保我没有重载ajaxSend来发送

  4. ruby - 使用 Ruby 通过 Outlook 发送消息的最简单方法是什么? - 2

    我的工作要求我为某些测试自动生成电子邮件。我一直在四处寻找,但未能找到可以快速实现的合理解决方案。它需要在outlook而不是其他邮件服务器中,因为我们有一些奇怪的身份验证规则,我们需要保存草稿而不是仅仅发送邮件的选项。显然win32ole可以做到这一点,但我找不到任何相当简单的例子。 最佳答案 假设存储了Outlook凭据并且您设置为自动登录到Outlook,WIN32OLE可以很好地完成此操作:require'win32ole'outlook=WIN32OLE.new('Outlook.Application')message=

  5. Ruby - 如何将消息长度表示为 2 个二进制字节 - 2

    我正在使用Ruby,我正在与一个网络端点通信,该端点在发送消息本身之前需要格式化“header”。header中的第一个字段必须是消息长度,它被定义为网络字节顺序中的2二进制字节消息长度。比如我的消息长度是1024。如何将1024表示为二进制双字节? 最佳答案 Ruby(以及Perl和Python等)中字节整理的标准工具是pack和unpack。ruby的packisinArray.您的长度应该是两个字节长,并且按网络字节顺序排列,这听起来像是n格式说明符的工作:n|Integer|16-bitunsigned,network(bi

  6. 华为OD机试用Python实现 -【明明的随机数】 2023Q1A - 2

    华为OD机试题本篇题目:明明的随机数题目输入描述输出描述:示例1输入输出说明代码编写思路最近更新的博客华为od2023|什么是华为od,od薪资待遇,od机试题清单华为OD机试真题大全,用Python解华为机试题|机试宝典【华为OD机试】全流程解析+经验分享,题型分享,防作弊指南华为o

  7. 基于C#实现简易绘图工具【100010177】 - 2

    C#实现简易绘图工具一.引言实验目的:通过制作窗体应用程序(C#画图软件),熟悉基本的窗体设计过程以及控件设计,事件处理等,熟悉使用C#的winform窗体进行绘图的基本步骤,对于面向对象编程有更加深刻的体会.Tutorial任务设计一个具有基本功能的画图软件**·包括简单的新建文件,保存,重新绘图等功能**·实现一些基本图形的绘制,包括铅笔和基本形状等,学习橡皮工具的创建**·设计一个合理舒适的UI界面**注明:你可能需要先了解一些关于winform窗体应用程序绘图的基本知识,以及关于GDI+类和结构的知识二.实验环境Windows系统下的visualstudio2017C#窗体应用程序三.

  8. MIMO-OFDM无线通信技术及MATLAB实现(1)无线信道:传播和衰落 - 2

     MIMO技术的优缺点优点通过下面三个增益来总体概括:阵列增益。阵列增益是指由于接收机通过对接收信号的相干合并而活得的平均SNR的提高。在发射机不知道信道信息的情况下,MIMO系统可以获得的阵列增益与接收天线数成正比复用增益。在采用空间复用方案的MIMO系统中,可以获得复用增益,即信道容量成倍增加。信道容量的增加与min(Nt,Nr)成正比分集增益。在采用空间分集方案的MIMO系统中,可以获得分集增益,即可靠性性能的改善。分集增益用独立衰落支路数来描述,即分集指数。在使用了空时编码的MIMO系统中,由于接收天线或发射天线之间的间距较远,可认为它们各自的大尺度衰落是相互独立的,因此分布式MIMO

  9. ruby-on-rails - 在 Flash 警报 Rails 3 中显示错误消息 - 2

    如果我在模型中设置验证消息validates:name,:presence=>{:message=>'Thenamecantbeblank.'}我如何让该消息显示在闪光警报中,这是我迄今为止尝试过的方法defcreate@message=Message.new(params[:message])if@message.valid?ContactMailer.send_mail(@message).deliverredirect_to(root_path,:notice=>"Thanksforyourmessage,Iwillbeintouchsoon")elseflash[:error]

  10. 【Java入门】使用Java实现文件夹的遍历 - 2

    遍历文件夹我们通常是使用递归进行操作,这种方式比较简单,也比较容易理解。本文为大家介绍另一种不使用递归的方式,由于没有使用递归,只用到了循环和集合,所以效率更高一些!一、使用递归遍历文件夹整体思路1、使用File封装初始目录,2、打印这个目录3、获取这个目录下所有的子文件和子目录的数组。4、遍历这个数组,取出每个File对象4-1、如果File是否是一个文件,打印4-2、否则就是一个目录,递归调用代码实现publicclassSearchFile{publicstaticvoidmain(String[]args){//初始目录Filedir=newFile("d:/Dev");Datebeg

随机推荐