【spring boot结合rabbit mq 到点执行，可精确到秒】

【spring boot结合rabbit mq 到点执行，可精确到秒】

  • 创建队列枚举
  • 创建自定义的队列消息pojo
  • 创建队列和延迟队列
  • 发送mq 消息
  • 接收mq 消息
  • DateTimeUtil
  • 测试
  • 注意点

创建队列枚举

public enum QueueEnum {/*** 各种异步消息频道*/TEST(1,"test","队列频道"),DELAY_TEST(2,"delay_test","延迟延迟频道"),;private Integer code;private String channel;private String desc;QueueEnum(Integer code, String channel, String desc) {this.code = code;this.channel = channel;this.desc = desc;}public Integer getCode() {return code;}public void setCode(Integer code) {this.code = code;}public String getChannel() {return channel;}public void setChannel(String channel) {this.channel = channel;}public String getDesc() {return desc;}public void setDesc(String desc) {this.desc = desc;}public static String findChannelByCode(Integer code) {QueueEnum[] queueEnums = QueueEnum.values();for (QueueEnum queueEnum : queueEnums) {if (code == queueEnum.getCode()) {return queueEnum.getChannel();}}return "";}
}

创建自定义的队列消息pojo


import java.io.Serializable;
import java.time.LocalDate;/**** 队列消息** 注意：涉及序列化问题，请勿将此类移动与修改* @author linjianhui*/
public class QueueMessage implements Serializable {private static final long serialVersionUID = 1L;//自定义的队列枚举private QueueEnum queueEnum;private String activityId;/*** 任务日期- yyyy-MM-dd* 任务日期- yyyy-MM-dd HH:mm:ss*/private String taskDate;private String msgId;public String getActivityId() {return activityId;}public String getTaskDate() {return taskDate==null? LocalDate.now().toString():taskDate;}public void setQueueEnum(QueueEnum queueEnum) {this.queueEnum = queueEnum;}public void setActivityId(String activityId) {this.activityId = activityId;}public void setTaskDate(String taskDate) {this.taskDate = taskDate;}public String getMsgId() {return msgId;}public void setMsgId(String msgId) {this.msgId = msgId;}public QueueEnum getQueueEnum() {return queueEnum;}public QueueMessage() {}public QueueMessage(QueueEnum queueEnum, String activityId) {this.queueEnum = queueEnum;this.activityId = activityId;}public QueueMessage(QueueEnum queueEnum, String activityId,String msgId) {this.queueEnum = queueEnum;this.activityId = activityId;this.msgId=msgId;}@Overridepublic String toString() {final StringBuilder sb = new StringBuilder("QueueMessage{");sb.append("queueEnum=").append(queueEnum);sb.append(", activityId='").append(activityId).append('\'');sb.append(", taskDate='").append(taskDate).append('\'');sb.append(", mgsId='").append(msgId).append('\'');sb.append('}');return sb.toString();}

创建队列和延迟队列

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;import java.util.HashMap;@Configuration
//保证队列的创建优先于监听队列
@Order(1)
public class TestRabbitConfig {@Bean("testQueue")public Queue testQueue() {return new Queue(QueueEnum.TEST.getChannel());}@Bean("testExchange")public DirectExchange testExchange() {return new DirectExchange(QueueEnum.TEST.getChannel());}/*** 将队列绑定到exchange,使用指定的路由key* @return*/@BeanBinding bindingtestQueueToExchange(@Qualifier("testQueue") Queue testQueue, @Qualifier("testExchange")DirectExchange testExchange) {return BindingBuilder.bind(testQueue).to(testExchange).with(QueueEnum.TEST.getChannel());}/*** 描述：定义延迟更新队列【死信队列】*  当队列到期后就会通过死信交换机和路由key，路由到指定队列* x-message-ttl 消息定时时间* x-max-length  队列最大长度* x-dead-letter-exchange：出现dead letter之后将dead letter重新发送到指定exchange* x-dead-letter-routing-key：出现dead letter之后将dead letter重新按照指定的routing-key发送* @param* @return*/@Bean("delayTestQueue")public Queue delayTestQueue() {HashMap<String, Object> arguments = new HashMap<>(4);//设置延15天// arguments.put("x-message-ttl", 15*24*6*10*60*1000);//需要时可以打开// x-message-ttl这个设置对队列中所有的消息有效【属于队列级别】//如果你想要【为每个消息动态设置过期时间】，你需要在【消息级别】设置Time To Live (TTL)。在Spring AMQP中，你可以通过设置MessageProperties的expiration属性来实现这一点：//在convertAndSend传入MessagePostProcessor实现类，覆盖其方法postProcessMessage(Message message)，使用message.getMessageProperties().setExpiration(delayInMs+"");来为消息单独设置过期时间// arguments.put("x-message-ttl", 10*60*1000);//10分钟arguments.put("x-max-length", 500000);arguments.put("x-dead-letter-exchange", QueueEnum.TEST.getChannel());arguments.put("x-dead-letter-routing-key", QueueEnum.TEST.getChannel());return new Queue(QueueEnum.DELAY_TEST.getChannel(), true, false, false, arguments);}/*** 描述：定义延迟更新队列交换机* @param* @return*/@Bean("delayTestExchange")public DirectExchange delayTestExchange() {return new DirectExchange(QueueEnum.DELAY_TEST.getChannel());}/*** 描述：绑定延迟更新队列到exchange* @param* @return*/@BeanBinding bindingDelayTestQueueToExchange(@Qualifier("delayTestQueue")Queue delayTestQueue, @Qualifier("delayTestExchange")DirectExchange delayTestExchange) {return BindingBuilder.bind(delayTestQueue).to(delayTestExchange).with(QueueEnum.DELAY_TEST.getChannel());}

发送mq 消息


import com.alibaba.fastjson.JSON;
import com.project.utils.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.time.Duration;
import java.time.LocalDateTime;/*** 描述：发送消息*/
@Component
@Slf4j(topic = "sendMqTask")
public class SendMqMessage {@AutowiredRabbitTemplate rabbitTemplate;public void sendTestMessage(QueueMessage queueMessage) {String messageId = StringUtil.getUniqueId("mq-");queueMessage.setMsgId(messageId);rabbitTemplate.convertAndSend(queueMessage.getQueueEnum().getChannel(), queueMessage.getQueueEnum().getChannel(), queueMessage, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {// 计算时间差long delayInMs = Duration.between(LocalDateTime.now(), DateTimeUtil.fromString2LocalDateTime(queueMessage.getTaskDate())).toMillis();//如果你想要为每个消息动态设置过期时间，你需要在【消息级别：更加细粒度控制】设置Time To Live (TTL)。在Spring AMQP中，你可以通过设置MessageProperties的expiration属性来实现这一点：//在convertAndSend传入MessagePostProcessor实现类，覆盖其方法postProcessMessage(Message message)，使用message.getMessageProperties().setExpiration(delayInMs+"");来为消息单独设置过期时间//这里，expiration属性的值是以毫秒为单位的过期时间戳。当这个时间戳过去后，消息就会变为死信//这样每条消息都有自己的过期时间，不用受死信队列的x-message-ttl的影响，死信队列的x-message-ttl这个设置对队列中所有的消息有效【队列级别】//在RabbitMQ中，如果同时在队列级别和消息级别设置了TTL（x-message-ttl 和 expiration 属性），那么将会遵循以下原则：// 1. 消息级别的TTL（expiration）优先：如果消息自身携带了TTL属性，那么即使队列设置了x-message-ttl，也会以消息本身的TTL为准。消息过期后，会被当作死信处理。// 2. 队列级别的TTL（x-message-ttl）作为默认值：只有当消息没有携带TTL属性时，才会使用队列级别的x-message-ttl作为消息的过期时间。// 因此，在你的场景中，如果同时设置了队列级别的x-message-ttl和消息级别的message.getMessageProperties().setExpiration(delayInMs+"")，那么将会以消息级别的TTL为准。//设置消息多长时间后过期message.getMessageProperties().setExpiration(delayInMs+"");return message;}});}
}

接收mq 消息

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.exceptions.PersistenceException;
import org.mybatis.spring.MyBatisSystemException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.annotation.Order;
import org.springframework.jdbc.CannotGetJdbcConnectionException;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;/*** 描述：消息消费监听*/
@Component
@Order(2)
@Slf4j(topic = "receiveMqTask")
public class ReceiveMqMessage {// private static final Logger MQ_LOG = LoggerFactory.getLogger("mqTask");@Value("${spring.profiles.active}")private String active;/*** 判断是否是正式环境** @return*/private boolean isProdEnv() {return "prod".equals(active);}/*** 判断是否是测试环境** @return*/private boolean isTestEnv() {return "test".equals(active);}/*** 监听消息队列* @param queueMessage* @param message : org.springframework.amqp.core.Message* @param channel : com.rabbitmq.client.Channel*/@RabbitListener(queues = ApiConstants.TEST)@RabbitHandlerpublic void test(QueueMessage queueMessage, Message message, Channel channel) {String env=isProdEnv()?"正式":isTestEnv()?"测试":active;log.info("====={}== test Mq Message={}",env, queueMessage);// String consumerTag = message.getMessageProperties().getConsumerTag();long deliveryTag = message.getMessageProperties().getDeliveryTag();try {System.out.println("发送时间是:"+ queueMessage.getTaskDate());System.out.println("当前时间是:"+ LocalDateTime.now().toLocalDate()+" "+LocalDateTime.now().toLocalTime());// 手动ACKtry {channel.basicAck(deliveryTag, false);} catch (IOException e) {log.error("MQ手动ACK错误： ", e);}} catch (Exception e) {log.error("test queue 失败");}}
}

DateTimeUtil

/*** 日期工具类*/
public class DateTimeUtil {/*** yyyy-MM-dd HH:mm:ss*/public static final String FORMAT_DATETIME = "yyyy-MM-dd HH:mm:ss";/*** discription: */public static String getLocalDateTime(LocalDateTime localDateTime) {DateTimeFormatter df = DateTimeFormatter.ofPattern(DateTimeUtil.FORMAT_DATETIME);if (localDateTime != null) {String localTime = df.format(localDateTime);return localTime;}return null;}
}

测试

@RestController
@RequestMapping(value = "/test")
public class TestController {@Autowiredprivate SendMqMessage sendMqMessage;@RequestMapping(value = "/testMqMessage", method = RequestMethod.GET)public ResultEntity testMqMessage(@RequestParam(value = "second",defaultValue = "20",required = false) Long second){QueueMessage queueMessage = new QueueMessage(QueueEnum.DELAY_TEST,"123");//设置20秒后更新【默认】queueMessage.setTaskDate(DateTimeUtil.getLocalDateTime(LocalDateTime.now().plusSeconds(second)));sendMqMessage.sendTestMessage(queueMessage);return "发送成功";}
}

注意点

//如果你想要为每个消息动态设置过期时间，你需要在【消息级别：更加细粒度控制】设置Time To Live (TTL)。在Spring AMQP中，你可以通过设置MessageProperties的expiration属性来实现这一点：//在convertAndSend传入MessagePostProcessor实现类，覆盖其方法postProcessMessage(Message message)，使用message.getMessageProperties().setExpiration(delayInMs+"");来为消息单独设置过期时间//这里，expiration属性的值是以毫秒为单位的过期时间戳。当这个时间戳过去后，消息就会变为死信//这样每条消息都有自己的过期时间，不用受死信队列的x-message-ttl的影响，死信队列的x-message-ttl这个设置对队列中所有的消息有效【队列级别】//在RabbitMQ中，如果同时在队列级别和消息级别设置了TTL（x-message-ttl 和 expiration 属性），那么将会遵循以下原则：// 1. 消息级别的TTL（expiration）优先：如果消息自身携带了TTL属性，那么即使队列设置了x-message-ttl，也会以消息本身的TTL为准。消息过期后，会被当作死信处理。// 2. 队列级别的TTL（x-message-ttl）作为默认值：只有当消息没有携带TTL属性时，才会使用队列级别的x-message-ttl作为消息的过期时间。// 因此，在你的场景中，如果同时设置了队列级别的x-message-ttl和消息级别的message.getMessageProperties().setExpiration(delayInMs+"")，那么将会以消息级别的TTL为准。

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

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

相关文章

等保2.0高风险项全解析：判定标准与应对方法

等保2.0高风险项全解析：判定标准与应对方法

引言 所谓高风险项&#xff0c;就是等保测评时可以一票否决的整改项&#xff0c;如果不改&#xff0c;无论你多少分都会被定为不合格。全文共58页&#xff0c;写得比较细了&#xff0c;但是想到大家基本不会有耐心去仔细看的&#xff08;凭直觉&#xff09;。这几天挑里边相对…
阅读更多...
android 网络请求总结

android 网络请求总结

1 先看下基础部分&#xff1a; android okhttp网络访问是基于 tcp/ip 的 最上层是应用层的封装&#xff0c;有http&#xff0c;https&#xff08;加密&#xff09;&#xff0c;ftp 下面是socket套接字的封装&#xff0c;就是将ip和端口的封装 在下面就是tcp/udp 在下面 ip协议…
阅读更多...
Java学习--学生管理系统(残破版)

Java学习--学生管理系统(残破版)

代码 Main.java import java.util.ArrayList; import java.util.Scanner;public class Main {public static void main(String[] args) {ArrayList<Student> list new ArrayList<>();loop:while (true) {System.out.println("-----欢迎来到阿宝院校学生管理系…
阅读更多...
可视化图文报表

可视化图文报表

Apache Echarts介绍 Apache Echarts是一款基于Javascript的数据可视化图表库&#xff0c;提供直观&#xff0c;生动&#xff0c;可交互&#xff0c;可个性化定制的数据可视化图表。 官网&#xff1a;Apache ECharts 入门案例&#xff1a; <!DOCTYPE html> <html>…
阅读更多...
flutter build ipa 打包比 xcode archive 打出的ipa包大

flutter build ipa 打包比 xcode archive 打出的ipa包大

为什么 flutter build ipa 打包比 xcode archive 打出的ipa包大&#xff1f; 如果你用Flutter构建的.ipa文件比通过Xcode Archive构建的.ipa文件要大&#xff0c;这可能是因为Flutter构建了一个包含了多平台的二进制文件的通用包。这意味着在Flutter构建的.ipa中包含了所有的C…
阅读更多...
C#，弗洛伊德-瑞文斯特（Floyd-Rivest）算法与源代码

C#，弗洛伊德-瑞文斯特（Floyd-Rivest）算法与源代码

Robert W. Floyd 1 Floyd-Rivest 算法 Floyd-Rivest 算法是一种选择算法&#xff0c;用于在不同元素的数组中找到第k个最小元素。它类似于快速选择算法&#xff0c;但在实际运行中有更好的运行时间。 和 QuickSelect 一样&#xff0c;该算法基于分区的思想工作。对数组进行分…
阅读更多...
济南适宜地提取

济南适宜地提取

题目： 网上下载中国的DEM、土地利用地图(1980、2000、2015年的）和一张最新济南市行政区划 图(要求:莱芜市并入济南后的区划图); 2.网上下载中国2015年年平均降水空间插值数据;3.．网上下载中国2015年年平均气温空间插值数据; (注:以上数据可到资源环境科学与数据中心下载http…
阅读更多...
51单片机 串口

51单片机 串口

一、串口基本认知概念 串口是一种用于在计算机或其他设备之间进行数据传输的通信接口。串口传输是通过发送和接收数据位来进行的&#xff0c;通常是一个字节一个字节地传输。串口通常有多种参数设置&#xff0c;比如波特率、数据位、校验位和停止位等&#xff0c;这些参数需要…
阅读更多...
服务器感染了.Elbie勒索病毒，如何确保数据文件完整恢复？

服务器感染了.Elbie勒索病毒，如何确保数据文件完整恢复？

导言&#xff1a; 今天将和大家分享一个备受关注的话题&#xff1a;Elbie勒索病毒。它不仅是数字世界中的恶梦&#xff0c;更是我们数据安全的宿敌。但是&#xff0c;别担心&#xff0c;我将为你揭开它的面纱&#xff0c;并提供如何挽救被它锁定的数据以及预防的方法。如果您正…
阅读更多...
【Docker】安装及相关的命令

【Docker】安装及相关的命令

目录 一 Docker简介 1.1 是什么 1.2 优缺点 1.3 应用场景 1.4 安装 二 命令 2.1 Docker基本命令 2.2 Docker镜像命令 2.3 Docker容器命令 一 Docker简介 1.1 是什么 Docker是一个开源的应用容器引擎&#xff0c;它基于Go语言实现&#xff0c;并利用操作系统本身已有的…
阅读更多...
【QT+QGIS跨平台编译】之五十四：【QGIS_CORE跨平台编译】—【qgssqlstatementlexer.cpp生成】

【QT+QGIS跨平台编译】之五十四：【QGIS_CORE跨平台编译】—【qgssqlstatementlexer.cpp生成】

文章目录 一、Flex二、生成来源三、构建过程一、Flex Flex (fast lexical analyser generator) 是 Lex 的另一个替代品。它经常和自由软件 Bison 语法分析器生成器 一起使用。Flex 最初由 Vern Paxson 于 1987 年用 C 语言写成。 “flex 是一个生成扫描器的工具，能够识别文本中…
阅读更多...
电机控制----电机极对数的作用

电机控制----电机极对数的作用

电机控制----电机极对数的作用 1. 电机极对数的概念和含义 电机的极对数是指电机运动部件磁场南、北极之间相邻的个数。简单来说&#xff0c;电机极对数就是电机磁极的数量。一般情况下&#xff0c;电机极对数越多&#xff0c;电机的转速越慢&#xff0c;但是转矩越大&#xf…
阅读更多...
状态机2——————

状态机2——————

1.原理 2.代码 2.1 complex_fsm.v module complex_fsm(input wire sys_clk ,input wire sys_rst_n ,input wire pi_money_half ,input wire pi_money_one ,output reg po_cola ,output reg po_money );parameter IDLE 5b00001,HALF 5b00010,ONE 5b00100…
阅读更多...
算法竞赛实用板子

算法竞赛实用板子

一、声明 自用版参考acwing&#xff0c;致力于实用、好用&#xff0c;板子中有个人理解&#xff0c;持续更新。 二、开板 1.快排 void quick_sort(int q[],int l,int r) {if(l>r)return; //出口int il-1,jr1,xq[lr>>1]; //分治方法while(i<j){do i;w…
阅读更多...
k8s（6）

k8s（6）

目录 一.kubectl 命令行管理K8S 陈述式资源管理方式&#xff08;可理解成使用一条kubectl命令及其参数选项来实现资源对象的管理操作&#xff09; service的4的基本类型&#xff1a; service的端口 应用发布策略&#xff1a; 声明式资源管理方式&#xff08;可理解成使用…
阅读更多...
spring boot 自定义注解封装(@RequestLimit注解)

spring boot 自定义注解封装(@RequestLimit注解)

描述&#xff1a; RequestLimit(time3)防重复点击&#xff0c;限制单个会话的请求频率。但是 RequestLimit注解中的参数不能传常量。 例如&#xff1a;不能通过获取application.yml文件中的配置信息&#xff0c;去动态配置限制时间&#xff0c;所以就对RequestLimit注解封装。 …
阅读更多...
Go语言反射(reflect)

Go语言反射(reflect)

反射是应用程序检查其所拥有的结构&#xff0c;尤其是类型的一种能。每种语言的反射模型都不同&#xff0c;并且有些语言根本不支持反射。Go语言实现了反射&#xff0c;反射机制就是在运行时动态调用对象的方法和属性&#xff0c;即可从运行时态的示例对象反求其编码阶段的定义…
阅读更多...
LeetCode_21_简单_合并两个有序链表

LeetCode_21_简单_合并两个有序链表

文章目录 1. 题目2. 思路及代码实现&#xff08;Python&#xff09;2.1 递归2.2 迭代 1. 题目 将两个升序链表合并为一个新的 升序 链表并返回。新链表是通过拼接给定的两个链表的所有节点组成的。 示例 1&#xff1a; 输入&#xff1a; l 1 [ 1 , 2 , 4 ] , l 2 [ 1 , 3 …
阅读更多...
MongoDB聚合运算符：$cmp

MongoDB聚合运算符：$cmp

文章目录 语法用法举例 $cmp聚合运算符返回连个值的比较结果。 语法 { $cmp: [ <expression1>, <expression2> ] }表达式可以是任何类型&#xff0c;使用标准的BSON比较顺序对不同类型的值进行比较。 用法 $cmp对两个值进行比较&#xff0c;返回&#xff1a; …
阅读更多...
SQL注入漏洞解析-less-8(布尔盲注)

SQL注入漏洞解析-less-8(布尔盲注)

我们来看一下第八关 当我们进行尝试时&#xff0c;他只有You are in...........或者没有显示。 他只有对和错显示&#xff0c;那我们只能用对或者错误来猜他这个数据库 ?id1%27%20and%20ascii(substr(database(),1,1))>114-- ?id1%27%20and%20ascii(substr(database(),1,…
阅读更多...
最新文章

Copyright @ 2022~2023