通过rabbitmq生成延时消息,并生成rabbitmq镜像

通过rabbitmq生成延时消息队列,并生成rabbitmq镜像

  • 整体描述
    • 1. 使用场景
    • 2. 目前问题
    • 3. 前期准备
  • 具体步骤
    • 1. 拉取镜像
    • 2. 运行镜像
    • 3. 安装插件
    • 4. 代码支持
      • 4.1 config文件
      • 4.2 消费监听
      • 4.2 消息生产
    • 5. 功能测试
  • 镜像操作
    • 1. 镜像制作
    • 2. 镜像导入
  • 总结

整体描述

1. 使用场景

在使用消息队列时,我们有时候需要生成一些延时消息,比如判断一个任务的开始时间,我在创建任务的时候计算出此时距离任务开始的时间,然后往消息队列里发送一个延时消息,我们希望等到任务开始的时候,再消费此消息,此时任务开始,可以进行一些业务上的操作。

2. 目前问题

之前写过一篇创建rabbitmq镜像的文章,链接: 在centos搭建rabbitmq并制作docker镜像,使用的rabbitmq的版本是3.6.8,只能通过过期时间expiration来设置消息的过期时间,在消息过期的时候,会进入死信队列中,也能达到上述要求。但是,但是,这个过期时间expiration,rabbitmq在处理的时候有个坑,前面消息如果没有过期,后面的消息就算过期了,也不会触发,就是先发的消息没有到期,之后再发的消息就算到期了,也不会触发回调。这显然不行。

3. 前期准备

需要准备的主要就是docker环境,这个可以自行搜一下怎么安装docker环境,由于和本文主要讲的内容关系不大,略…

具体步骤

为了解决此问题,我们可以用延时队列插件来实现,这个插件时一个开发者写的,在github上但是已经被rabbitmq官方接受了,所以可以放心用。

1. 拉取镜像

首先我们先拉取一个rabbitmq的官方镜像进行操作,这个需要注意一下拉取的版本,由于延时队列的插件支持的版本是3.7之后的rabbitmq,所以需要拉取3.7之后的,我这拉取的是3.8.17版本。在命令行输入:

docker pull rabbitmq:3.8.17-management

注:这个如果报错,看下自己的docker环境有没有问题。带management是带管理页面的镜像,我们选用的带management的镜像,后期使用的时候好操作和定位问题。

2. 运行镜像

拉取成功之后,使用命令:

docker images

查看镜像是否拉取成功,如下就是成功了:
rabbitmq镜像
之前用的3.6.8的,不支持延时消息队列的插件…
然后运行镜像,创建容器并启动:

docker run --name rabbitmq-server -p 5672:5672 -p 15672:15672 -d rabbitmq:3.8.17-management

此时用:

docker ps -a

查看容器:
rabbitmq容器
容器已经创建并启动,我们通过web页面可以访问rabbitmq的管理页面,在浏览器输入:http://localhost:15672/
默认账号:guest,密码:guest
rabbitmq登录页面

3. 安装插件

此时rabbitmq已经运行,我们需要安装插件来支持延时消息队列,插件下载地址
选择相应的rabbitmq版本进行下载,注意版本不要选错了。下载完是一个rabbitmq_delayed_message_exchange-3.8.0.ez的文件,我们需要把这个文件上传到docker的/opt/rabbitmq/plugins目录下。
上传之后,进入/opt/rabbitmq/sbin目录执行如下命令让插件生效:

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

执行之后看到如下就成功了:
插件启动成功
成功之后刷新一下管理页面,在新建交换机那里,type能多一个x-delayed-message的选项:
添加延时交换机
此时,我们的rabbitmq就配置完成了。

4. 代码支持

rabbitmq目前已经可以接收延时消息了,在代码端我们也需要进行相应的修改,以达到发送延时消息的目的。

4.1 config文件

package com.thcb.rabbitmq.config;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;/*** RabbitMQ的配置类** @author thcb* @date 2023-09-05*/
@Configuration
public class RabbitMqConfig {// 交换机private static final String DELAYED_EXCHANGE = "delayed.exchange";// 队列private static final String DELAYED_QUEUE = "delayed.queue";// 路由private static final String DELAYED_ROUTING_KEY = "delayed.routingKey";/*** 队列*/@Beanpublic Queue delayedQueue() {return new Queue(DELAYED_QUEUE);}/*** 交换机*/@Beanpublic CustomExchange delayedExchange() {Map<String, Object> args = new HashMap<>();args.put("x-delayed-type", "direct");return new CustomExchange(DELAYED_EXCHANGE, "x-delayed-message", false, false, args);}/*** 绑定延迟队列和交换机*/@Beanpublic Binding delayQueueBindingDelayExchange() {return BindingBuilder.bind(delayedQueue()).to(delayedExchange()).with(DELAYED_ROUTING_KEY).noargs();}}

4.2 消费监听

package com.thcb.rabbitmq.recevier;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.Date;/*** 消费监听** @author thcb* @date 2023-09-05*/
@Slf4j
@Component
public class DelayQueueReceiver {@RabbitListener(queues = "delayed.queue")public void receiveDelayedQueue(Message message) {String msg = new String(message.getBody());log.info("当前时间:{},收到DelayedQueue消息:{}", new Date().toString(), msg);}}

4.2 消息生产

这里创建一个controller来生产消息,里面有两个接口,一个生产消息的延时时间是5秒,另一个是30秒,用来测试延时时间。

package com.thcb.rabbitmq.controller;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;import java.util.Date;/*** 消息生产controller** @author thcb* @date 2023-09-05*/
@RestController
@RequestMapping("/HelloController")
public class HelloController {private static final Logger log = LoggerFactory.getLogger(HelloController.class);@Autowiredprivate AmqpTemplate rabbitTemplate;@RequestMapping("/sendXDLMessage1")@ResponseBodypublic String sendXDLMessage1() {int time = 5000;String message = "{\"type\":\"sendXDLMessage1\"}";log.info("当前时间:{},发送一条延迟{}毫秒的信息给延迟队列:{}", new Date().toString(), time, message);rabbitTemplate.convertAndSend("delayed.exchange", "delayed.routingKey", message, msg -> {msg.getMessageProperties().setDelay(time);return msg;});return "sendXDLMessage1 success";}@RequestMapping("/sendXDLMessage2")@ResponseBodypublic String sendXDLMessage2() {int time = 30000;String message = "{\"type\":\"sendXDLMessage2\"}";log.info("当前时间:{},发送一条延迟{}毫秒的信息给延迟队列:{}", new Date().toString(), time, message);rabbitTemplate.convertAndSend("delayed.exchange", "delayed.routingKey", message, msg -> {msg.getMessageProperties().setDelay(time);return msg;});return "sendXDLMessage2 success";}
}

5. 功能测试

代码修改完,就可以测试了,启动工程之后,在rabbitmq管理页面能看到自动创建了如下交换机和队列:
创建的交换机
创建的队列
可以看到交换机的类型是x-delayed-message。
接下来就可以调用测试接口,生产2条消息看看了。先调用sendXDLMessage2接口,生产一个延时30秒的消息,过一会再调用sendXDLMessage1的接口,生产一个延时5秒的消息。log结果如下:
运行结果
结果符合我们的预期,先发的30秒延时消息消息2,之后发的5秒延时消息1,然后过了5秒消息1先回调,之后30秒消息2回调。

镜像操作

使用docker主要就是要制作镜像,之后直接就可以用了要不每次还得配置。提示制作之前,把现在的队列和交换机都删除,队列和交换机是通过代码创建的,账号可以换一个,默认的guest不太安全。
一切都准备就绪,就可以制作镜像了。

1. 镜像制作

将镜像打包成tar文件。

docker commit 【镜像id】 rabbitmq:3.8.17
docker save -o rabbitmq-3.8.17.tar rabbitmq:3.8.17

2. 镜像导入

制作完镜像进行导入

docker load <rabbitmq-3.8.17.tar
docker run -d -p 5672:5672 -p 15672:15672 --privileged --restart=always --name rabbitmq rabbitmq:3.8.17

总结

以上就是rabbitmq延时消息的相关内容,另外这个延时消息在消息很多的情况下可能会有一些性能问题,使用的时候需要注意一下。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/73712.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Stable Diffusion WebUI中COMMANDLINE_ARGS参数配置说明

Windows用户: 在webui-user.bat文件的set COMMANDLINE_ARGS=后面配置 Linux用户: 在webui-user.sh文件的export COMMANDLINE_ARGS=""引号中配置 COMMANDLINE_ARGS详细参数配置说明 配置(Configuration)-h, –help无错误显示此帮助消息并退出–config配置配置/稳…

蠕虫病毒流量分析案例

背景 某供排水集团的网络管理员对其网络的健康状况持认可态度&#xff0c;表示网络运行正常&#xff0c;没有发现异常行为。然而&#xff0c;由于网络环境变得越来越复杂&#xff0c;仅凭借传统的网络经验已经不能全面了解网络情况。因此&#xff0c;我们为供排水集团安装了Ne…

Codeforces Round 895 (Div. 3)

A. Two Vessels 求出目标水位&#xff0c;然后计算倒水的次数向上取整即可 void solve(){db a, b, c; cin >> a >> b >> c;print(int(ceil((max(a, b) - (abs(b a) / 2)) / c))); } B. The Corridor or There and Back Again 对于每个可能被激活的陷阱&…

使用Docker部署Gitlab的记录

docker版本 使用docker -v查看 Docker version 1.13.1, build 7d71120/1.13.1运行容器镜像 映射本机的9980端口为Docker内部的80端口 映射本机的9922端口为Docker内部的22端口 使用root用户启动 映射本机目录/mnt/sda/gitlab/log为Docker内部的/var/log/gitlab 映射本机目录…

4 | Java Spark实现 WordCount

简单的 Java Spark 实现 WordCount 的教程,它将教您如何使用 Apache Spark 来统计文本文件中每个单词的出现次数。 首先,确保您已经安装了 Apache Spark 并设置了运行环境。您需要准备一个包含文本内容的文本文件,以便对其进行 WordCount 分析。 代码 package com.bigdat…

我的创作纪念日-2023

突然看到CSDN有这么一个发文模版&#xff08;后文都是按照模版填空了&#xff09;&#xff0c;那就写一些吧。 机缘 具体有什么机缘呢&#xff1f;简单就是说&#xff0c;在CSDN上得到了一些帮助&#xff0c;因此也想写一些文章以期帮助别人。为什么不在自己的网站写呢&#…

Golang复习

golang的特点 Golang 针对并发进行了优化&#xff0c;并且在规模上运行良好 自动垃圾收集明显比 Java 或 Python 更有效&#xff0c;因为它与程序同时执行 golang数据类型 基本数据类型&#xff08;值类型&#xff09; 布尔类型 数字类型 整型 根据有符号分为&#xff1a;…

[NLP]LLM---FineTune自己的Llama2模型

一 数据集准备 Let’s talk a bit about the parameters we can tune here. First, we want to load a llama-2-7b-hf model and train it on the mlabonne/guanaco-llama2-1k (1,000 samples), which will produce our fine-tuned model llama-2-7b-miniguanaco. If you’re …

axios 请求和响应拦截器

1. 创建实例 使用 axios.create() 使用自定义配置创建一个 axios 实例。 const $http axios.create({timeout: 1000,headers: {Content-Type: application/json,} })2. 拦截器 在请求或响应被 then 或者 catch 处理前拦截他们&#xff0c;拦截分为请求拦截和响应拦截。 //…

微信小程序电影票订票小程序软件设计与实现

摘 要 我们的生活水平正在不断的提高&#xff0c;然而提高的一个重要的侧面表现就是更加注重我们的娱乐生活。电影是我们都喜欢的一种娱乐方式&#xff0c;各式各样的电影给我们带来的喜悦也是大不相同的。带来快乐的同时也因为其复杂、繁琐的流程让电影爱好者们变得烦躁起来。…

以太坊虚拟机

1.概述 以太坊虚拟机 EVM 是智能合约的运行环境。它不仅是沙盒封装的&#xff0c;而且是完全隔离的&#xff0c;也就是说在 EVM 中运行代码是无法访问网络、文件系统和其他进程的。甚至智能合约之间的访问也是受限的。 2.账户 以太坊中有两类账户&#xff08;它们共用同一个…

人大金仓分析型数据库备份和恢复(三)

目录 备份或恢复过滤内容 过滤叶子分区 使用gprestore过滤 备份或恢复过滤内容 gpbackup备份指定数据库中的所有schema和表&#xff0c;除非您使用schema级别或表级别过滤器选项排除或包含单个schema或表对象。 schema级别选项是--include-schema或--exclude-schema命令行选…

​@Cacheable 注解​

1. 功能说明 Cacheable 注解在方法上&#xff0c;表示该方法的返回结果是可以缓存的。也就是说&#xff0c;该方法的返回结果会放在缓存中&#xff0c;以便于以后使用相同的参数调用该方法时&#xff0c;会返回缓存中的值&#xff0c;而不会实际执行该方法。 注意&#xff0c;这…

华为云API对话机器人CBS的魅力—实现简单的对话操作

云服务、API、SDK&#xff0c;调试&#xff0c;查看&#xff0c;我都行 阅读短文您可以学习到&#xff1a;人工智能AI智能的问答管理、全面的对话管理、高效训练部署 1.IntelliJ IDEA 之API插件介绍 API插件支持 VS Code IDE、IntelliJ IDEA等平台、以及华为云自研 CodeArts …

Unity 2018发布在iOS 16.3偶尔出现画面不动的问题

1&#xff09;Unity 2018发布在iOS 16.3偶尔出现画面不动的问题 2&#xff09;IL2CPP在Xcode下增量编译问题 3&#xff09;帧同步实现PuppetMaster布娃娃系统的问题 这是第351篇UWA技术知识分享的推送&#xff0c;精选了UWA社区的热门话题&#xff0c;涵盖了UWA问答、社区帖子等…

每日刷题-3

目录 一、选择题 二、编程题 1、计算糖果 2、进制转换 一、选择题 1、 解析&#xff1a;在C语言中&#xff0c;以0开头的整数常量是八进制的&#xff0c;而不是十进制的。所以&#xff0c;0123的八进制表示相当于83的十进制表示&#xff0c;而123的十进制表示不变。printf函数…

ASP.NET Core IOC容器

//IOC容器支持依赖注入{ServiceCollection serviceDescriptors new ServiceCollection();serviceDescriptors.AddTransient<IMicrophone, Microphone>();serviceDescriptors.AddTransient<IPower, Power>();serviceDescriptors.AddTransient<IHeadphone, Headp…

【SQL应知应会】索引 • Oracle版:B-树索引;位图索引;函数索引;单列与复合索引;分区索引

欢迎来到爱书不爱输的程序猿的博客, 本博客致力于知识分享,与更多的人进行学习交流 本文免费学习,自发文起3天后,会收录于SQL应知应会专栏,本专栏主要用于记录对于数据库的一些学习,有基础也有进阶,有MySQL也有Oracle 索引 • MySQL版 前言一、Oracle索引1.索引概述及分类…

upload-labs1-17思路

1 直接写一个php文件测试一下&#xff0c;发现弹窗不让上传 原理很简单&#xff0c;就是把后缀名拿出来过滤一遍&#xff0c;而白名单就是弹窗的这三个 解决方法&#xff1a; 因为这是在前端防御的一个手段&#xff0c;所以直接在浏览器设置上禁用js就行了&#xff1a; 也可…

springboot整合neo4j--采用Neo4jClient和Neo4jTemplate方式

1.背景 看了spring-boot-starter-data-neo4j的源码之后发现&#xff0c;该starter内已经实现了Neo4jClient和Neo4jTemplate&#xff0c;我们只需要使用Autowire就能直接使用它操作neo4j。 Neo4jClient方式与我的另一篇springboot整合neo4j-使用原生cypher Java API博客方式一样…