当前位置：首页 > news >正文

Java实现Redis延时队列

news 来源：原创 2024/5/8 12:38:29

“如何实现Redis延时队列”这个面试题应该也是比较常见的，解答如下：
使用sortedset（有序集合），拿时间戳作为 score ，消息内容作为key 调用 zadd 来生产消息，消费者用zrangebyscore 指令获取 N 秒之前的数据轮询进行处理。

目录

实现思路
引入Jedis
指令简介
zadd
zrem
zrangeByScore

Java实现Redis延时队列

实现思路

Java实现Redis延时队列，首先要了解何为延时队列，即可以将消息存储在队列中，并在指定的延时时间后再将消息出队。这种队列在很多场景下都非常有用，例如消息延时处理，延时确认（订单确认） 等，参考以上解答，思路应该拆分：
首先需要有个延时队列，该队列是通过一定顺序（当前时间戳+延时时间）排序的（即优先取到延时时间已结束的数据），然后消费者端就需要获取到队列中延时时间靠前结束的数据（即当前时间戳+延时时间靠前）。

引入Jedis

<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>3.7.0</version>
</dependency>

指令简介

zadd

zadd命令用于将一个成员Score值加入到有序集合中。Score值可以是整数或者浮点数。如果有序集合中已经存在相同的成员，那么旧成员将被替代。
语法：ZADD key Score member [Score2 member2 …]
示例：ZADD students 100 alice 或 ZADD students 80 alice 90 bob （添加单个或多个情况）

zrem

zrem命令用于从有序集合中移除一个或多个成员。该命令接收两个参数：第一个参数是要操作的有序集合的键，第二个参数是将要移除的成员的值。
语法：ZREM key member [member …]
示例：ZREM students alice

zrangeByScore

zrangeByScore命令用于获取分数在指定范围内的所有成员。
语法：ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count]
参数说明：
key：Redis中的键。
min：最小分数。
max：最大分数。
WITHSCORES：可选参数，如果设置为true，则返回分数以及成员。
LIMIT：可选参数，用于限制返回的成员数量。
offset：偏移量，从第几个成员开始。
count：限制返回的成员数量。
返回值：
按照分数升序返回成员列表。
如果设置了LIMIT参数，则返回限制数量的成员。
示例：ZADD ZRANGEBYSCORE students 80 90 WITHSCORES

Java实现Redis延时队列

核心部分，消息队列工具类

import redis.clients.jedis.Jedis;
import java.util.Set;public class DelayQueueWithRedis {private Jedis jedis;private String queueKey;public DelayQueueWithRedis(Jedis jedis, String queueKey) {this.jedis = jedis;this.queueKey = queueKey;}// 添加消息到延迟队列public void push(String message, long delaySeconds) {// 计算消息的分数，这里使用消息进入队列的时间加上延迟时间long score = System.currentTimeMillis() / 1000 + delaySeconds;//向有序集合添加一个成员，并设置其分数jedis.zadd(queueKey, score, message);}// 获取并消费一条消息public String pop() {while (true) {long now = System.currentTimeMillis() / 1000;// 只获取分数在0到当前时间的元素Set<String> messages = jedis.zrangeByScore(queueKey, 0, now, 0, 1);if (messages.isEmpty()) {System.out.println("No messages");// 没有可消费的消息，休眠一会儿继续尝试try {Thread.sleep(500);} catch (InterruptedException e) {Thread.currentThread().interrupt();break;}} else {String message = messages.iterator().next();// 从有序集合中移除一个成员jedis.zrem(queueKey, message);return message;}}return null;}
}

生产者端测试

import redis.clients.jedis.Jedis;/*** @Author: zhangximing* @Email: 530659058@qq.com* @Date: 2024/2/19 16:53* @Description: 生产者端测试*/
public class MainP {public static void main(String[] args) {Jedis jedis = new Jedis("localhost",6379);DelayQueueWithRedis delayQueue = new DelayQueueWithRedis(jedis, "delay_queue");// 添加延时消息delayQueue.push("message1", 5);delayQueue.push("message2", 10);delayQueue.push("message3", 8);}}

消费者端测试

import redis.clients.jedis.Jedis;/*** @Author: zhangximing* @Email: 530659058@qq.com* @Date: 2024/2/19 16:51* @Description: 消费者端测试*/
public class MainC {public static void main(String[] args) {Jedis jedis = new Jedis("localhost",6379);DelayQueueWithRedis delayQueue = new DelayQueueWithRedis(jedis, "delay_queue");// 消费延时消息while (true) {String message = delayQueue.pop();if (message != null) {System.out.println("Consumed: " + message);}}}
}