Java实现Redis延时队列

news2024/11/23 17:08:20

“如何实现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);
            }
        }
    }
}