【Redis】redis高阶-使用zset实现延时队列

2024-06-03 22:20

本文主要是介绍【Redis】redis高阶-使用zset实现延时队列,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

Hi,大家好,我是抢老婆酸奶的小肥仔。

最近在使用redis时,就想能不能用其实现消息队列?也在网上看了下其他小伙伴写的实现,结合自身业务实现了如下消息队列,希望对大家有用。

废话不多说,直接开撸。

1、为什么zset可以做消息队列?

首先我们来看下,设计消息队列需要考虑的需求:有序性,消息重复性,可靠性。

  • 有序性:zset所有元素可以根据成员关联的score来进行从低到高的排序,例如,我们可以利用时间戳来进行排序
  • 消息重复性:在zset中每个元素都是唯一的,这也保证了消息的唯一性
  • 可靠性:zset会自动维护元素之间的顺序,在添加或删除元素时无需手动排序,提升操作速度。

2、使用的zset命令

命令描述
zadd将一个给定score的成员添加到有序集合中,返回添加元素的个数
zrange根据元素在有序排序中的位置,从有序集合中获取多个元素
rank(K key, Object o)获取指定元素在集合中的索引,索引从0开始

3、代码实现

使用zset实现消息队列时,具体的流程,如下:

生产者流程:

  1. 用户获取消息Id,并封装消息体
  2. 用户发送数据到生产者,先获取锁
  3. 如果获取到锁,则校验该消息体是否已添加到队列中,已添加则直接返回提醒。
  4. 若未添加则调用方法将数据保存到zset集合中,否则等到指定时间后再获取锁。
  5. 推送数据后,释放锁

消费者流程:

  1. 调用方法获取数据
  2. 获取到数据,则直接返回,否则到指定时间后再次获取数据,直到获取到数据并返回。

统一返回类:

/*** @Author: jiangjs* @Description:* @Date: 2021/11/12 15:46**/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ResultUtil<T> implements Serializable {private int code;private String msg;private T data;public static <T> ResultUtil<T> success(){return ResultUtil.<T>builder().code(1000).msg("成功").build();}public static <T> ResultUtil<T> success(T data){return ResultUtil.<T>builder().code(1000).msg("成功").data(data).build();}public static <T> ResultUtil<T> error(String msg){return ResultUtil.<T>builder().code(5000).msg(msg).data(null).build();}public static <T> ResultUtil<T> error(int code,String msg){return ResultUtil.<T>builder().code(code).msg(msg).build();}
}

3.1 消息实体

需添加消息Id,主要防止消息重复消费。

/*** @author: jiangjs* @description: 消息实体* @date: 2023/5/30 11:11**/
@Data
@Accessors(chain = true)
public class QueueTask<T> {/*** 消息Id*/private String taskId;/*** 任务*/private T task;
}

3.2 队列类型

队列类型可以理解为队列的名称,通过枚举,可以随意添加队列名称。

/*** @author: jiangjs* @description: 队列类型* @date: 2023/5/30 10:53**/
public enum QueueTypeEnum {/*** 订单*/ORDER("order");private final String type;QueueTypeEnum(String type){this.type = type;}public String getType(){return type;}
}

3.3 创建消息工具

package com.jiashn.springbootproject.redis.utils;import com.jiashn.springbootproject.redis.domain.QueueTask;
import com.jiashn.springbootproject.utils.ResultUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;/*** @author: jiangjs* @description: redis实现消息队列* @date: 2023/5/30 10:51**/
public class RedisQueueUtil<T> {private static final Logger log = LoggerFactory.getLogger(RedisQueueUtil.class);private RedisTemplate<String,QueueTask<T>> redisTemplate;/*** 队列类型,即名称*/private final QueueTypeEnum typeEnum;public RedisQueueUtil(QueueTypeEnum typeEnum,RedisTemplate<String,QueueTask<T>> redisTemplate){this.typeEnum = typeEnum;this.redisTemplate = redisTemplate;}/*** 添加消息数据* @param queueTask 消息* @param time 延迟时间,单位s*/public ResultUtil<String> sendQueueTask(QueueTask<T> queueTask, long time){//加锁if (getLock()){try {Long rank = redisTemplate.opsForZSet().rank(typeEnum.getType(), queueTask);if (Objects.nonNull(rank)){return ResultUtil.error(6000,"消息数据已经存在,不予添加......");}Boolean result = redisTemplate.opsForZSet().add(typeEnum.getType(), queueTask, System.currentTimeMillis() + time*1000);if (Objects.nonNull(result) && result){log.info("添加消息数据成功:" + queueTask + ",添加时间:" + LocalDateTime.now());return ResultUtil.success("添加消息数据成功");}return ResultUtil.error("添加消息数据失败");}finally {//释放锁releaseLock();}} else {log.info("未获取到锁,稍后再试");return ResultUtil.error("未获取到锁,稍后再试");}}/*** 获取zset前count数据* @param count 数据数* @return 返回获取到数据*/public Set<QueueTask<T>> loopGetTask(int count) {//rangeByScore,根据score顺序获取zset数据的值return redisTemplate.opsForZSet().rangeByScore(typeEnum.getType(), 0, System.currentTimeMillis(), 0, count-1);}/*** 注销消息队列* @param typeEnum 消息队列名称*/public void destroy(QueueTypeEnum typeEnum){redisTemplate.opsForZSet().remove(typeEnum.getType());}/*** 获取任务Id* @return 返回消息Id*/public String getTaskId(){return typeEnum.getType() + "_" + UUID.randomUUID().toString().replace("-","");}/*** 获取锁* @return 返回加锁状态*/private boolean getLock(){Boolean absent = redisTemplate.opsForValue().setIfAbsent(typeEnum.getType() + "_Locked", null, 30L, TimeUnit.MINUTES);return Objects.nonNull(absent) ? absent : false;}/*** 释放锁*/public void releaseLock(){redisTemplate.delete(typeEnum.getType() + "_Locked");}
}

在消息工具类中,创建消息任务时添加了锁,只有在获取锁的前提下才能添加消息任务。

提供获取消息Id的方法是为了让提交消息任务前,先获取Id,即使在提交时网络发生问题,提交的Id还是同一个,再进行消息消费时,可以根据这个Id来进行判断该消息任务是否已被消费,被消费则直接丢弃。

3.4 消费消息

/*** @author: jiangjs* @description: 启动消费* @date: 2023/5/30 14:27**/
@Component
public class CustomerTaskLineRunner implements CommandLineRunner {@Resourceprivate RedisTemplate<String,QueueTask<String>> redisTemplate;private final static String QUEUE_TYPE = QueueTypeEnum.ORDER.getType();private final static Logger log = LoggerFactory.getLogger(CustomerTaskLineRunner.class);@Overridepublic void run(String... args) throws Exception {RedisQueueUtil<String> queueUtil = new RedisQueueUtil<>(QueueTypeEnum.ORDER,redisTemplate);while (true){Set<QueueTask<String>> queueTasks = queueUtil.loopGetTask(10);if (CollectionUtils.isNotEmpty(queueTasks)){for (QueueTask<String> queueTask : queueTasks) {//校验当前消息是否已消费,主要防止网络延时,导致多次提交同一任务 存在QueueTask<String> stringQueueTask = redisTemplate.opsForValue().get(QUEUE_TYPE + "_" + queueTask.getTaskId());if (Objects.nonNull(stringQueueTask)){log.info("该任务已经消费,不能重复消费");redisTemplate.opsForZSet().remove(QUEUE_TYPE,queueTask);continue;}Long removeNum = redisTemplate.opsForZSet().remove(QUEUE_TYPE,queueTask);if (Objects.nonNull(removeNum) && removeNum > 0){String task = queueTask.getTask();log.info("消费任务数据:" + task);//设置过期时间,10分钟内则默认是重复提交redisTemplate.opsForValue().set(QUEUE_TYPE + "_" + queueTask.getTaskId(),queueTask,10L, TimeUnit.MINUTES);}}}log.info("------1分钟后再次获取------");Thread.sleep(60000);}}
}

校验重复消息,若消息重复且在10分钟内未被消费,则直接将该消息从队列中删除。在消息任务被消费后,将数据从队列中移除。

执行结果:

谢谢大家,今天的分享就到这,不合理的地方希望大家指正。

这篇关于【Redis】redis高阶-使用zset实现延时队列的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1028278

相关文章

中文分词jieba库的使用与实景应用(一)

知识星球:https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义: 精确模式(默认模式): 全模式: 搜索引擎模式: paddle 模式(基于深度学习的分词模式): 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取

使用SecondaryNameNode恢复NameNode的数据

1)需求: NameNode进程挂了并且存储的数据也丢失了,如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2)故障模拟 (1)kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 (2)删除NameNode存储的数据(/opt/module/hadoop-3.1.4/data/tmp/dfs/na

Hadoop数据压缩使用介绍

一、压缩原则 (1)运算密集型的Job,少用压缩 (2)IO密集型的Job,多用压缩 二、压缩算法比较 三、压缩位置选择 四、压缩参数配置 1)为了支持多种压缩/解压缩算法,Hadoop引入了编码/解码器 2)要在Hadoop中启用压缩,可以配置如下参数

Makefile简明使用教程

文章目录 规则makefile文件的基本语法:加在命令前的特殊符号:.PHONY伪目标: Makefilev1 直观写法v2 加上中间过程v3 伪目标v4 变量 make 选项-f-n-C Make 是一种流行的构建工具,常用于将源代码转换成可执行文件或者其他形式的输出文件(如库文件、文档等)。Make 可以自动化地执行编译、链接等一系列操作。 规则 makefile文件

hdu1043(八数码问题,广搜 + hash(实现状态压缩) )

利用康拓展开将一个排列映射成一个自然数,然后就变成了普通的广搜题。 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<queue>#include<map>#include<stdio.h>#include<stdlib.h>#include<ctype.h>#inclu

hdu1180(广搜+优先队列)

此题要求最少到达目标点T的最短时间,所以我选择了广度优先搜索,并且要用到优先队列。 另外此题注意点较多,比如说可以在某个点停留,我wa了好多两次,就是因为忽略了这一点,然后参考了大神的思想,然后经过反复修改才AC的 这是我的代码 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<

使用opencv优化图片(画面变清晰)

文章目录 需求影响照片清晰度的因素 实现降噪测试代码 锐化空间锐化Unsharp Masking频率域锐化对比测试 对比度增强常用算法对比测试 需求 对图像进行优化,使其看起来更清晰,同时保持尺寸不变,通常涉及到图像处理技术如锐化、降噪、对比度增强等 影响照片清晰度的因素 影响照片清晰度的因素有很多,主要可以从以下几个方面来分析 1. 拍摄设备 相机传感器:相机传

【C++】_list常用方法解析及模拟实现

相信自己的力量,只要对自己始终保持信心,尽自己最大努力去完成任何事,就算事情最终结果是失败了,努力了也不留遗憾。💓💓💓 目录   ✨说在前面 🍋知识点一:什么是list? •🌰1.list的定义 •🌰2.list的基本特性 •🌰3.常用接口介绍 🍋知识点二:list常用接口 •🌰1.默认成员函数 🔥构造函数(⭐) 🔥析构函数 •🌰2.list对象

【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

✨✨ 欢迎大家来到景天科技苑✨✨ 🎈🎈 养成好习惯,先赞后看哦~🎈🎈 🏆 作者简介:景天科技苑 🏆《头衔》:大厂架构师,华为云开发者社区专家博主,阿里云开发者社区专家博主,CSDN全栈领域优质创作者,掘金优秀博主,51CTO博客专家等。 🏆《博客》:Python全栈,前后端开发,小程序开发,人工智能,js逆向,App逆向,网络系统安全,数据分析,Django,fastapi

让树莓派智能语音助手实现定时提醒功能

最初的时候是想直接在rasa 的chatbot上实现,因为rasa本身是带有remindschedule模块的。不过经过一番折腾后,忽然发现,chatbot上实现的定时,语音助手不一定会有响应。因为,我目前语音助手的代码设置了长时间无应答会结束对话,这样一来,chatbot定时提醒的触发就不会被语音助手获悉。那怎么让语音助手也具有定时提醒功能呢? 我最后选择的方法是用threading.Time