Redis实现延时队列

缓存队列延时向接口报工,并支持多实例部署。

引入依赖

<dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-data</artifactId><version>3.17.4</version>
</dependency>

注入RedisClient

import org.springframework.util.StringUtils;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RedissonConfig {@Value("${spring.redis.host}")private String host;@Value("${spring.redis.port}")private String port;@Value("${spring.redis.password}")private String password;@Value("${spring.redis.database}")private Integer database;@Bean()public RedissonClient redissonClient() {Config config = new Config();SingleServerConfig serversConfig = config.useSingleServer();serversConfig.setAddress("redis://" + host + ":" + port);if (!StringUtils.isEmpty(password)){serversConfig.setPassword(password);}serversConfig.setDatabase(database);return Redisson.create(config);}
}

注入延时队列Bean

import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** @author* redisson延迟队列*/
@Configuration
public class RedissonQueueConfig {@Beanpublic RBlockingQueue<String> rBlockingQueue(@Qualifier("redissonClient") RedissonClient redissonClient) {String queueName = "queue";return redissonClient.getBlockingQueue(queueName);}@Bean(name = "rDelayedQueue")public RDelayedQueue<String> rDelayedQueue(@Qualifier("redissonClient") RedissonClient redissonClient,@Qualifier("rBlockingQueue") RBlockingQueue<String> blockQueue) {return redissonClient.getDelayedQueue(blockQueue);}
}

编写方法

import java.util.concurrent.TimeUnit;/*** @author*/
public interface DelayQueue {/*** 发布** @param object* @return*/Boolean offer(Object object);/*** 带延迟功能的队列** @param object* @param time* @param timeUnit*/void offer(Object object, Long time, TimeUnit timeUnit);void offerAsync(Object object, Long time, TimeUnit timeUnit);Boolean offerAsync(Object object);
}import org.redisson.api.RDelayedQueue;
import org.redisson.api.RFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;/*** @author*/
@Component
public class RedissonDelayQueue implements DelayQueue {private static Logger log = LoggerFactory.getLogger(RedissonDelayQueue.class);@Resource(name = "rDelayedQueue")private RDelayedQueue<Object> rDelayedQueue;@Overridepublic Boolean offer(Object object) {return rDelayedQueue.offer(object);}@Overridepublic void offer(Object object, Long time, TimeUnit timeUnit) {rDelayedQueue.offer(object, time, timeUnit);}@Overridepublic void offerAsync(Object object, Long time, TimeUnit timeUnit) {rDelayedQueue.offerAsync(object, time, timeUnit);}@Overridepublic Boolean offerAsync(Object object) {boolean flag = false;RFuture<Boolean> rFuture = rDelayedQueue.offerAsync(object);try {flag = rFuture.get();} catch (InterruptedException | ExecutionException e) {log.info("offerAsync exception:{}", e.getMessage());e.printStackTrace();}return flag;}
}

延时任务

队列生产

@Resource(name = "rDelayedQueue")
private RDelayedQueue<Object> rDelayedQueue;public void delayedQueue(){//rDelayedQueue.size()队列中元素数量//添加到延时队列中rDelayedQueue.offerAsync("延时30秒执行任务",30, TimeUnit.SECONDS);
}

队列消费

import com.coctrl.mom.common.service.WebService;
import com.coctrl.mom.process.entity.vo.LESPassStationDetailQueueVO;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import javax.annotation.Resource;/*** @author*/
@Component
@Slf4j
public class RedissonTask {@Resource(name = "rBlockingQueue")private RBlockingQueue<Object> rBlockingQueue;@PostConstructpublic void take() {new Thread(() -> {while (true) {try {log.info("延时报工信息===============" +rBlockingQueue.take());//业务代码} catch (InterruptedException e) {log.error("延时报工失败===============" + e.getMessage());e.printStackTrace();}}}).start();}
}

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

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

相关文章

flask中实现restful-api

flask中实现restful-api 举例&#xff0c;我们可以创建一个用于管理任务&#xff08;Task&#xff09;的API。在这个例子中&#xff0c;我们将有以下API&#xff1a; GET /tasks: 获取所有任务POST /tasks: 创建一个新的任务GET /tasks/<id>: 获取一个任务的详情PUT /t…

prometheus+grafana进行服务器资源监控

在性能测试中&#xff0c;服务器资源是值得关注一项内容&#xff0c;目前&#xff0c;市面上已经有很多的服务器资 源监控方法和各种不同的监控工具&#xff0c;方便在各个项目中使用。 但是&#xff0c;在性能测试中&#xff0c;究竟哪些指标值得被关注呢&#xff1f; 监控有…

c++ | const 对常量的绑定

这一块&#xff0c;感觉很真的不好描述。最好就是搬着primer去啃去理解了。 const 核心思想&#xff0c;就是const对象很多操作都被限制了。类型决定对象操作 const 常量引用&#xff0c;所引用的对象必须是确定的&#xff08;粗俗的讲就是字面值 常量&#xff09; const int …

appium自动爬取数据

爬取类容&#xff1a;推荐知识点中所有的题目 爬取方式&#xff1a;appium模拟操作获取前端数据 入门级简单实现&#xff0c;针对题目和答案是文字内容的没有提取出来 适用场景;数据不多&#xff0c;参数加密&#xff0c;反爬严格等场景 from appium import webdriver impor…

git 常用命令有哪些

Git 是我们开发工作中使用频率极高的工具&#xff0c;下面总结下他的基本指令有哪些&#xff0c;顺便温习一下。 前言 一般项目中长存2个分支&#xff1a; 主分支&#xff08;master&#xff09; 和开发分支&#xff08;develp&#xff09; 项目存在三种短期分支 &#xff1a…

Linux安装MySQL 8.1.0

MySQL是一个流行的开源关系型数据库管理系统&#xff0c;本教程将向您展示如何在Linux系统上安装MySQL 8.1.0版本。请按照以下步骤进行操作&#xff1a; 1. 下载MySQL安装包 首先&#xff0c;从MySQL官方网站或镜像站点下载MySQL 8.1.0的压缩包mysql-8.1.0-linux-glibc2.28-x…

k8s实战

一、pod 1. 配置文件 2.探针 【1】类型 StartupProde&#xff1a;启动探针&#xff0c;判断程序是否启动LivenessProbeReadlinessProbe&#xff1a; 确认应用是否启动成功 【2】探测方式 ExecAction&#xff1a;在容器内执行一个命令&#xff0c;如果返回值为0&#xff0…

快速WordPress个人博客并内网穿透发布到互联网

快速WordPress个人博客并内网穿透发布到互联网 文章目录 快速WordPress个人博客并内网穿透发布到互联网 我们能够通过cpolar完整的搭建起一个属于自己的网站&#xff0c;并且通过cpolar建立的数据隧道&#xff0c;从而让我们存放在本地电脑上的网站&#xff0c;能够为公众互联网…

[RK3568 Android11] 开发之默认授予系统预置的APP权限

[RK3568 Android11] 开发之默认授予系统预置的APP权限 [RK3568 Android11] 开发之默认授予系统预置的APP权限_rk3568 授予所有文件的管理权限_~未来可期~的博客-CSDN博客

分享 一个类似 ps 辅助线功能

效果图片&#xff1a; 提示&#xff1a;这里的样式我做边做了修改&#xff0c;根据个人情况而定。 //你也可以npm下载 $ npm install --save vue-ruler-tool特点 没有依赖可拖动的辅助线快捷键支持 开始使用 1. even.js /*** description 绑定事件 on(element, event, han…

通用商城项目(中)

金山编译器出问题了。下面段落标号全出问题了&#xff0c;排版也出问题了。懒得改了。 使用对象存储OSS&#xff0c;保存品牌logo 新建Module&#xff0c;提供上传、显示服务 有些不明所以的&#xff0c;按照steinliving-commodity配置了一通pom.xml 新建application.yml&…

【NLP概念源和流】 06-编码器-解码器模型(6/20 部分)

一、说明 在机器翻译等任务中,我们必须从一系列输入词映射到一系列输出词。读者必须注意,这与“序列标记”不同,在“序列标记”中,该任务是将序列中的每个单词映射到预定义的类,如词性或命名实体任务。 作者生成 在上面的

【嵌入式学习笔记】嵌入式入门3——串口

1.数据通信的基础概念 1.1.串行/并行通信 数据通信按数据通信方式分类&#xff1a;串行通信、并行通信 1.2.单工/半双工/全双工通信 数据通信按数据传输方向分类&#xff1a;单工通信、半双工通信、全双工通信 单工通信&#xff1a;数据只能沿一个方向传输半双工通信&…

【vue学习】watch监听问题

watch监听 执行的时候&#xff0c;watch方法未执行。并没有报错 遇见的问题 子组件&#xff0c;定义了props&#xff0c;属性里有一个selectedData&#xff0c;但是在执行的时候&#xff0c;这个对象在父组件里也修改了&#xff0c;可是就是死活没有触发。。。 观察报错 比…

防雷工程行业应用和施工工艺

防雷工程是指通过各种手段和措施&#xff0c;保护建筑物、设备和人员免受雷电侵害的技术。在我国&#xff0c;由于雷电活动频繁&#xff0c;防雷工程的重要性不言而喻。地凯科技将介绍防雷工程的基本知识、相关案例以及防雷器产品。 一、防雷工程的基本知识 雷电的危害 雷电…

真机搭建中小网络

这是b站上的一个视频&#xff0c;演示了如何搭建一个典型的中小网络&#xff0c;供企业使用 一、上行端口&#xff1a;上行端口就是连接汇聚或者核心层的口&#xff0c;或者是出广域网互联网的口。也可理解成上传数据的端口。 二、下行端口&#xff1a;连接数据线进行下载的端…

pytorch学习——如何构建一个神经网络——以手写数字识别为例

目录 一.概念介绍 1.1神经网络核心组件 1.2神经网络结构示意图 1.3使用pytorch构建神经网络的主要工具 二、实现手写数字识别 2.1环境 2.2主要步骤 2.3神经网络结构 2.4准备数据 2.4.1导入模块 2.4.2定义一些超参数 2.4.3下载数据并对数据进行预处理 2.4.4可视化数…

生成模型相关算法:EM算法步骤和公式推导

EM算法 引言EM算法例子及解法EM算法步骤和说明 引言 EM 算法是一种选代算法&#xff0c;1977 年 Dempster 等人总结提出&#xff0c;用于含有隐变量(hidden variable)的概率模型参数的极大似然估计&#xff0c;或极大后验概率估计EM算法的每次选代由两步组成:E步&#xff0c;求…

编写docker-compose和docker file支持vim命令

下面是一个简单的docker-compose.yml文件示例&#xff0c;其中包含一个运行Ubuntu镜像的容器&#xff0c;并且已经安装了vim编辑器。 version: 3 services:mycontainer:image: ubuntucommand: tail -f /dev/nulltty: truestdin_open: truevolumes:- ./mydata:/datacontainer_n…

RocketMQ生产者和消费者都开启Message Trace后,Consume Message Trace没有消费轨迹

一、依赖 <dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.0.3</version> </dependency>二、场景 1、生产者和消费者所属同一个程序 2、生产者开启消…