Redis使用ZSET实现消息队列的项目实践

import redis.clients.jedis.Jedis;
import java.util.Set;
public class RedisMessageQueue {
    private Jedis jedis; //Redis连接对象
    private String queueName; //队列名字
    /**
     * 构造函数
     * @param host Redis主机地址
     * @param port Redis端口
     * @param password Redis密码
     * @param queueName 队列名字
     */
    public RedisMessageQueue(String host, int port, String password, String queueName){
        jedis = new Jedis(host, port);
        jedis.auth(password);
        this.queueName = queueName;
    }
    /**
     * 发送消息
     * @param message 消息内容
     */
    public void sendMessage(String message){
        //获取当前时间戳
        long timestamp = System.currentTimeMillis();
        //将消息添加到有序集合中
        jedis.zadd(queueName, timestamp, message);
    }
    /**
     * 接收消息
     * @param count 一次接收的消息数量
     * @return 返回接收到的消息
     */
    public String[] receiveMessage(int count){
        //设置最大轮询时间
        long timeout = 5000;
        //获取当前时间戳
        long start = System.currentTimeMillis();
        while (true) {
            //获取可用的消息数量
            long size = jedis.zcount(queueName, "-inf", "+inf");
            if (size == 0) {
                //如果无消息，休眠50ms后继续轮询
                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                //计算需要获取的消息数量count与当前可用的消息数量size的最小值
                count = (int) Math.min(count, size);
                //获取消息
                Set<String> messages = jedis.zrange(queueName, 0, count - 1);
                String[] results = messages.toArray(new String[0]);
                //移除已处理的消息
                jedis.zremrangeByRank(queueName, 0, count - 1);
                return results;
            }
            //检查是否超时
            if (System.currentTimeMillis() - start > timeout) {
                return null; //超时返回空
            }
        }
    }
    /**
     * 销毁队列
     */
    public void destroy(){
        jedis.del(queueName);
        jedis.close();
    }
}

public static void main(String[] args) {
    //创建消息队列
    RedisMessageQueue messageQueue = new RedisMessageQueue("localhost", 6379, "password", "my_queue");
    //生产者发送消息
    messageQueue.sendMessage("message1");
    messageQueue.sendMessage("message2");
    //消费者接收消息
    String[] messages = messageQueue.receiveMessage(10);
    System.out.println(Arrays.toString(messages)); //输出：[message1, message2]
    //销毁队列
    messageQueue.destroy();
}

# 获取 zset 集合中所有元素
ZREVRANGE queue +inf -inf WITHSCORES
# 获取 zset 集合中第1到第10个元素（分数从大到小排列）
ZREVRANGE queue +inf -inf WITHSCORES LIMIT 0 9
# 获取 zset 集合中分数在 1581095012 到当前时间之间的元素
ZREVRANGEBYSCORE queue +inf 1581095012 WITHSCORES