Redis消息队列、阻塞队列、延时队列的实现

更新时间：2023年11月10日 11:28:07 作者：夏寻.

Redis是一种常用的内存数据库,它提供了丰富的功能,通常用于数据缓存和分布式队列,本文主要介绍了Redis消息队列、阻塞队列、延时队列的实现,感兴趣的可以了解一下

一、普通队列

可以直接使用Redis的list数据类型实现消息队列，只需简单的两个指令lpush和rpop或者rpush和lpop。

lpush+rpop：左进右出的队列
rpush+lpop：左出右进的队列

使用redis的命令来模拟普通队列

使用lpush命令生产消息：

>lpush queue:single 1
"1"
>lpush queue:single 2
"2"
>lpush queue:single 3
"3"

使用rpop命令消费消息：

>rpop queue:single
"1"
>rpop queue:single
"2"
>rpop queue:single
"3"

使用Java代码来实现普通队列：

生产者SingleProducer

package com.morris.redis.demo.queue.single;

import redis.clients.jedis.Jedis;

/**
 * 生产者
 */
public class SingleProducer {

    public static final String SINGLE_QUEUE_NAME = "queue:single";

    public static void main(String[] args) {
        Jedis jedis = new Jedis();
        for (int i = 0; i < 100; i++) {
            jedis.lpush(SINGLE_QUEUE_NAME, "hello " + i);
        }
        jedis.close();
    }
}

消费者SingleConsumer：

package com.morris.redis.demo.queue.single;

import redis.clients.jedis.Jedis;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 消费者
 */
public class SingleConsumer {

    public static void main(String[] args) throws InterruptedException {
        Jedis jedis = new Jedis();
        while (true) {
            String message = jedis.rpop(SingleProducer.SINGLE_QUEUE_NAME);
            if(Objects.nonNull(message)) {
                System.out.println(message);
            } else {
                TimeUnit.MILLISECONDS.sleep(500);
            }
        }
    }
}

上面的代码已经基本实现了普通队列的生产与消费，但是上述的例子中消息的消费者存在两个问题：

普通的redis队列，为了实现业务，通常会使用while进行循环，这样的话没有消息时依旧会频繁的执行循环，造成cpu的空转，所以一般会在代码中增加sleep来解决该问题，但因此又会造成消息延迟问题。
如果生产者速度大于消费者消费速度，消息队列长度会一直增大，时间久了会占用大量内存空间。

二、Redis阻塞队列

redis队列提供了 “阻塞式” 拉取消息的命令：BRPOP / BLPOP，这里的 B 指的是阻塞（Block）。
如果队列为空，消费者在拉取消息时就「阻塞等待」，一旦有新消息过来，就通知消费者立即处理新消息。

使用redis的brpop命令来模拟阻塞队列

>brpop queue:single 30

使用 BRPOP 这种阻塞式方式拉取消息时，还支持传入一个「超时时间」，如果设置为 0，则表示不设置超时，直到有新消息才返回，否则会在指定的超时时间后返回 NULL

Java阻塞队列生产者实现如下：

package com.cxh;

import org.junit.jupiter.api.Test;
import redis.clients.jedis.Jedis;

import java.util.concurrent.TimeUnit;

/**
 * 生产者类
 * 生产者每隔600ms生成一条消息
 * */
class MessageProducer extends Thread{
    public static final String MESSAGE_KEY = "message:queue";
    private volatile int count;

    public void putMessage(String mess){
        Jedis jedis = new Jedis("127.0.0.1", 6379);
/*        jedis.auth("123456");*/
        Long size = jedis.lpush(MESSAGE_KEY, mess);
        System.out.println("Put " + Thread.currentThread().getName() + " put message " + count);
        count++;
    }
    @Override
    public synchronized  void run() {
        for(int i = 0 ; i < 1; i++){
            putMessage("message" + count);
            try {
                Thread.sleep(600);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

Java阻塞队列消费者实现如下：

package com.cxh.Component;

import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;

import java.util.Arrays;
import java.util.List;

@Component
public class component implements CommandLineRunner {
    public static final String MESSAGE_KEY = "message:queue";
    @Override
    public void run(String... args) throws Exception {
/*        //todo: 需要执行的方法
        System.out.println(Arrays.toString(args));*/
        System.out.println("comsumer 111");
        Jedis jedis = new Jedis("127.0.0.1", 6379);
//        String message = jedis.rpop(MESSAGE_KEY);
//        System.out.println("Pop " + Thread.currentThread().getName() + "comsumer message = " + message); //多线程的时候使用
        while (true){
            List<String> message = jedis.brpop(0, MESSAGE_KEY);
            System.out.println(message.toString());
        }
    }

}

上面的代码已经实现了延迟队列的生产与消费，需要注意的是：

无法实现一次生产多次消费（使用 pub/sub 发布订阅模式，可以实现 1:N 的消息队列，即一次生产，多端消费）
阻塞时间结束后代码会继续向下执行
如果设置的超时时间太长，这个连接太久没有活跃过，可能会被 Redis Server 判定为无效连接，之后 Redis Server 会强制把这个客户端踢下线。所以，客户端要有处理机制。
实际项目中redis连接超时时间远大于20s，因此正常情况不会出现redis超时问题。以防万一增加redis异常捕获，出现异常时杀掉当前进程，同时supervisord会自动重新拉起该进程

三、Redis延迟队列

zset 会按 score 进行排序，如果 score 代表想要执行时间的时间戳。在某个时间将它插入zset集合中，它会按照时间戳大小进行排序，也就是对执行时间前后进行排序。

起一个死循环线程不断地进行取第一个key值，如果当前时间戳大于等于该key值的socre就将它取出来进行消费删除，可以达到延时执行的目的。

下面使用redis的zset来模拟延时队列

命令生产者：

127.0.0.1:6379> zadd queue:delay 1 order1 2 order2 3 order3
(integer) 0

命令消费者：

127.0.0.1:6379> zrange queue:delay 0 0 withscores
1) "order1"
2) "1"
127.0.0.1:6379> zrem queue:delay order1
(integer) 1

使用Java代码来实现普通队列：

生产者DelayProducer ：

package com.morris.redis.demo.queue.delay;

import redis.clients.jedis.Jedis;

import java.util.Date;
import java.util.Random;

/**
 * 生产者
 */
public class DelayProducer {

    public static final String DELAY_QUEUE_NAME = "queue:delay";

    public static void main(String[] args) {
        Jedis jedis = new Jedis();
        long now = new Date().getTime();
        Random random = new Random();
        for (int i = 0; i < 10; i++) {
            int second = random.nextInt(30); // 随机订单失效时间
            jedis.zadd(DELAY_QUEUE_NAME, now + second * 1000, "order"+i);
        }
        jedis.close();
    }
}

消费者DelayConsumer ：

package com.morris.redis.demo.queue.delay;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Tuple;

import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * 消费者
 */
public class DelayConsumer {

    public static void main(String[] args) throws InterruptedException {
        Jedis jedis = new Jedis();
        while (true) {
            long now = new Date().getTime();
            Set<Tuple> tupleSet = jedis.zrangeWithScores(DelayProducer.DELAY_QUEUE_NAME, 0, 0);
            if(tupleSet.isEmpty()) {
                TimeUnit.MILLISECONDS.sleep(500);
            } else {
                for (Tuple tuple : tupleSet) {
                    Double score = tuple.getScore();
                    long time = score.longValue();
                    if(time < now) {
                        jedis.zrem(DelayProducer.DELAY_QUEUE_NAME, tuple.getElement());
                        System.out.println("order[" + tuple.getElement() +"] is timeout at " + time);
                    } else {
                        TimeUnit.MILLISECONDS.sleep(500);
                    }
                    break;
                }
            }
        }
    }
}

应用场景

延时队列在项目中的应用还是比较多的，尤其像电商类平台：

12306 下单成功后，在半个小时内没有支付，自动取消订单。
如果订单一直处于某一个未完结状态时，及时处理关单，并退还库存。
淘宝新建商户一个月内还没上传商品信息，将冻结商铺等。
会议预定系统，在预定会议开始前半小时通知所有预定该会议的用户。
安全工单超过 24 小时未处理，则自动拉企业群提醒相关责任人。
用户下单外卖以后，距离超时时间还有 10 分钟时提醒外卖小哥即将超时。
外卖平台发送订餐通知，下单成功后 60s 给用户推送短信。

到此这篇关于Redis消息队列、阻塞队列、延时队列的实现的文章就介绍到这了,更多相关Redis消息队列、阻塞队列、延时队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

基于Redis实现基本抢红包算法详解
[key, value]的缓存数据库, Redis官方性能描述非常高, 所以面对高并发场景, 使用Redis来克服高并发压力是一个不错的手段, 本文主要基于Redis来实现基本的抢红包系统设计,感兴趣的朋友跟随小编一起看看吧
2024-04-04
Linux安装Redis、后台运行、系统自启动的设置方法
Redis是用C语言编写的开源免费的高性能的分布式内存数据库，基于内存运行并支持持久化的NoSQL数据库。这篇文章主要介绍了Linux安装Redis、后台运行、系统自启动,需要的朋友可以参考下
2020-01-01
Redis分布式锁的实现方式
本文主要介绍了Redis分布式锁的实现方式，分布式锁是满足分布式系统或集群模式下多进程可见并且互斥的锁。感兴趣的同学可以参考阅读
2023-04-04
Redis安装教程图解
Redis是一个开源的使用ANSI C语言编写、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库，并提供多种语言的API。本文就教大家如何安装Redis，需要的朋友可以参考下
2015-10-10
华为欧拉openEuler编译安装Redis的实现步骤
本文主要介绍了华为欧拉openEuler编译安装Redis的实现步骤，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-01-01
深入了解Redis的看门狗机制
Redis锁的延期机制,通常被称为看门狗机制,本文就拉介绍一下Redis的看门狗机制,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
2024-12-12
CentOS Linux系统下安装Redis过程和配置参数说明
这篇文章主要介绍了CentOS Linux系统下安装Redis过程和配置参数说明,需要的朋友可以参考下
2014-10-10
Redis中的zset的底层实现过程
这篇文章主要介绍了Redis中的zset的底层实现过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2025-06-06
Redis数组和链表深入详解
这篇文章主要介绍了Redis数组和链表深入详解，这是redis的基础的知识点，有感兴趣的同学可以学习下
2021-03-03
Redis中的Hash和List类型常用命令及应用分析
Redis内存数据库中,除了基础的String类型外,Hash和List也是开发中频繁使用的两种数据结构,下面详细讲解Hash和List类型的特点、常用命令及典型应用场景,以便处理复杂业务需求,需要的朋友跟随小编一起看看吧
2025-08-08