Redis 做延迟消息队列

news2025/7/22 12:49:21

背景

看到消息队列,我们肯定会想到各种MQ,比如:RabbitMQ,acivityMQ、RocketMQ、Kafka等。

但是,当我们需要使用消息中间件的时候,并非每次都需要非常专业的消息中间件,假如我们只有一个消息队列,只有一个消费者,那就没有必要去使用上面这些专业的消息中间件,这种情况我们可以考虑使用 Redis 来做消息队列。

延迟消息队列使用场景

  1. 我们打车,在规定时间内,没有车主接单,那么平台就会推送消息给你,提示暂时没有车主接单。
  2. 网上支付场景,下单了,如果没有在规定时间付款,平台通常会发消息提示订单在有效期内没有支付完成,此订单自动取消之类的信息。
  3. 我们买东西,如果在一定时间内,没有对该订单进行评分,这时候平台会给一个默认分数给此订单。

.....

Redis如何实现消息队列?

大家都知道,Redis的五种数据类型,其中有一种类型是list。并且提供了相应的进入list的命令lpush和rpush ,以及弹出list的命令lpop和rpop。

这里我们就可以把List理解为一个消息队列,并且lpush和rpush操作称之为进入队列,同时,lpop和rpop称之为消息出队列。

命令

lpush

  • 将一个或多个值 value 插入到列表 key 的表头
  • 如果有多个 value 值,那么各个 value 值按从左到右的顺序依次插入到表头:比如说,对空列表 mylist 执行命令 LPUSH mylist a b c ,列表的值将是 c b a ,
  • 这等同于原子性地执行 LPUSH mylist a 、 LPUSH mylist b 和 LPUSH mylist c 三个命令。
  • 如果 key 不存在,一个空列表会被创建并执行 LPUSH 操作。
  • 当 key 存在但不是列表类型时,返回一个错误。

使用案例:

rpush

  • 将一个或多个值 value 插入到列表 key 的表尾(最右边)。
  • 如果有多个 value 值,那么各个 value 值按从左到右的顺序依次插入到表尾:比如对一个空列表 mylist 执行 RPUSH mylist a b c ,得出的结果列表为 a b c ,等同于执行命令 RPUSH mylist a 、 RPUSH mylist b 、 RPUSH mylist c 。
  • 如果 key 不存在,一个空列表会被创建并执行 RPUSH操作。
  • 当 key 存在但不是列表类型时,返回一个错误。

使用案例:

lpop

使用方式:lpop key。移除并返回列表 key 的头元素。如果key不存在,返回nil。

使用案例:

 

rpop

使用方式:rpop key,移除并返回列表 key 的尾元素。当 key 不存在时,返回 nil 。

使用案例:

 

以上四个命令是不是相当的简单呢,这里说一下lrange命令。

lrange

  • 返回列表 key 中指定区间内的元素,区间以偏移量 start 和 stop 指定。
  • 下标(index)参数 start 和 stop 都以 0 为底,也就是说,以 0 表示列表的第一个元素,以 1 表示列表的第二个元素,以此类推。
  • 你也可以使用负数下标,以 -1 表示列表的最后一个元素, -2 表示列表的倒数第二个元素,以此类推。

Redis实现消息队列

使用Spring Boot+Redis实现:

添加application.properties内容:

# Redis数据库索引(默认为0)
spring.redis.database=0
# Redis服务器地址
spring.redis.host=127.0.0.1
# Redis服务器连接端口
spring.redis.port=6379
# Redis服务器连接密码(默认为空)
spring.redis.password=
# 连接池最大连接数(使用负值表示没有限制)
spring.redis.jedis.pool.max-active=20
# 连接池最大阻塞等待时间(使用负值表示没有限制)
spring.redis.jedis.pool.max-wait=-1
# 连接池中的最大空闲连接
spring.redis.jedis.pool.max-idle=10
# 连接池中的最小空闲连接
spring.redis.jedis.pool.min-idle=0
# 连接超时时间(毫秒)
spring.redis.timeout=1000

pom.xml中添加:

<dependency>
   <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
 </dependency>

创建一个RedisConfig类,对RedisTemplate做一些序列化的设置:

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class RedisConfig {

    @Autowired
    private RedisConnectionFactory redisConnectionFactory;

    @Bean
    public RedisTemplate<String, Object> redisTemplate() {
        RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
        template.setConnectionFactory(redisConnectionFactory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new StringRedisSerializer());
        template.afterPropertiesSet();
        return template;
    }
}

创建RedisMQServicehe RedisMQServiceImpl

public interface RedisMQService {
    void produce(String string);
    void consume();
}
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class RedisMQServiceImpl implements RedisMQService {
    private static Logger log = LoggerFactory.getLogger(RedisMQServiceImpl.class);

    private static final String MESSAGE_KEY = "message:queue";

    @Resource
    private RedisTemplate redisTemplate;

    @Override
    public void produce(String string) {
        //生产者把消息丢到消息队列中
        redisTemplate.opsForList().leftPush(MESSAGE_KEY, string);
    }

    @Override
    public void consume() {
        String string = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY);
        //消费方拿到消息后进行业务处理
        log.info("consume : {}", string);
    }
}

创建一个ccontroller

import com.tian.user.mq.RedisMQService;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@RestController
public class RedisMQController {
    @Resource
    private RedisMQService redisMQService;

    @PostMapping("/produce")
    public String produce() {
        String[] names = {"java后端技术全栈", "老田", "面试专栏"};
        for (String name : names) {
            redisMQService.produce(name);
        }
        return "ok";
    }

    @PostMapping("/consume")
    public void consume() {
        int i = 0;
        while (i < 3) {
            redisMQService.consume();
            i++;
        }
    }
}

启动项目,访问:http://localhost:8080/produce生产者把三个消息丢到消息队列中。

在访问:http://localhost:8080/consume从消息队列中取出消息,然后就可以拿着消息继续做相关业务了。

后台输出:

 

到此,使用Redis实现消息队列就成功了。

但是,搞了半天只是使用Redis实现 了消息队列,那延迟呢?

上面并没有提到延迟队列的实现方式,下面我们来看看Redis中是如何实现此功能的。

Redis实现延迟消息队列的相关命令

延迟队列可以通过 zset 来实现,因为 zset 中有一个 score,我们可以把时间作为 score,将 value 存到 redis 中,然后通过轮询的方式,去不断的读取消息出来

整体思路

1.消息体设置有效期,设置好score,然后放入zset中

2.通过排名拉取消息

3.有效期到了,就把当前消息从zset中移除

我们来看看,zset有哪些命令:

 

可以通过网址:http://doc.redisfans.com/set/index.html 获取中文版Redis命令。

其实Redis实现延迟队列,只需要zset的三个命令即可。下面先来熟悉这三个命令。

zadd命令

使用方式:ZADD key score member [[score member][score member] ...]

  • 将一个或多个 member 元素及其 score 值加入到有序集 key 当中。
  • 如果某个 member 已经是有序集的成员,那么更新这个 member 的 score 值,并通过重新插入这个 member 元素,来保证该 member 在正确的位置上。
  • score 值可以是整数值或双精度浮点数。
  • 如果 key 不存在,则创建一个空的有序集并执行 ZADD 操作。
  • 当 key 存在但不是有序集类型时,返回一个错误。
  • 在 Redis 2.4 版本以前, ZADD 每次只能添加一个元素。

使用案例:

ZRANGEBYSCORE命令

使用方式:ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count]

1.返回有序集 key 中,所有 score 值介于 min 和 max 之间(包括等于 min 或 max )的成员。有序集成员按 score 值递增(从小到大)次序排列。

2.具有相同 score 值的成员按字典序(lexicographical order)来排列(该属性是有序集提供的,不需要额外的计算)。

3.可选的 LIMIT 参数指定返回结果的数量及区间(就像SQL中的 SELECT LIMIT offset, count ),注意当 offset 很大时,定位 offset 的操作可能需要遍历整个有序集,此过程最坏复杂度为 O(N) 时间。

4.可选的 WITHSCORES 参数决定结果集是单单返回有序集的成员,还是将有序集成员及其 score 值一起返回。

注意:该选项自 Redis 2.0 版本起可用。

区间及无限

min 和 max 可以是 -inf 和 +inf ,这样一来,你就可以在不知道有序集的最低和最高 score 值的情况下,使用 ZRANGEBYSCORE这类命令。

默认情况下,区间的取值使用闭区间(小于等于或大于等于),你也可以通过给参数前增加 ( 符号来使用可选的开区间 (小于或大于)。

举个例子:

ZRANGEBYSCORE zset (1 5

返回所有符合条件 1 < score <= 5 的成员,而

ZRANGEBYSCORE zset (5 (10

则返回所有符合条件 5 < score < 10 的成员。

使用案例

ZREM命令

使用方式:ZREM key member [member ...]

移除有序集 key 中的一个或多个成员,不存在的成员将被忽略。

当 key 存在但不是有序集类型时,返回一个错误。

使用案例:

延迟队列可以通过Zset(有序列表实现),Zset类似于java中SortedSet和HashMap的结合体,它是一个Set结构,保证了内部value值的唯一,同时他还可以给每个value设置一个score作为排序权重,Redis会根据score自动排序,我们每次拿到的就是最先需要被消费的消息,利用这个特性我们可以很好实现延迟队列。

java代码实现

创建一个消息实体类:

import java.time.LocalDateTime;

public class Message {
    /**
     * 消息唯一标识
     */
    private String id;
    /**
     * 消息渠道 如 订单 支付 代表不同业务类型
     * 为消费时不同类去处理
     */
    private String channel;
    /**
     * 具体消息 json
     */
    private String body;

    /**
     * 延时时间 被消费时间  取当前时间戳+延迟时间
     */
    private Long delayTime;

    /**
     * 创建时间
     */
    private LocalDateTime createTime;
    // set get 省略
}

生产者方代码:

import com.tian.user.dto.Message;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.UUID;

@Component
public class MessageProvider {

    @Resource
    private DelayingQueueService delayingQueueService;

    private static String USER_CHANNEL = "USER_CHANNEL";

    /**
     * 发送消息
     *
     * @param messageContent
     */
    public void sendMessage(String messageContent, long delay) {
        try {
            if (messageContent != null) {
                String seqId = UUID.randomUUID().toString();
                Message message = new Message();
                //时间戳默认为毫秒 延迟5s即为 5*1000
                long time = System.currentTimeMillis();
                LocalDateTime dateTime = Instant.ofEpochMilli(time).atZone(ZoneOffset.ofHours(8)).toLocalDateTime();
                message.setDelayTime(time + (delay * 1000));
                message.setCreateTime(dateTime);
                message.setBody(messageContent);
                message.setId(seqId);
                message.setChannel(USER_CHANNEL);
                delayingQueueService.push(message);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

消费方代码:

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.tian.user.dto.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.List;

@Component
public class MessageConsumer {
    private static ObjectMapper mapper = Jackson2ObjectMapperBuilder.json().build();
    @Resource
    private DelayingQueueService delayingQueueService;
    private Logger log = LoggerFactory.getLogger(this.getClass());

    /**
     * 定时消费队列中的数据
     * zset会对score进行排序 让最早消费的数据位于最前
     * 拿最前的数据跟当前时间比较 时间到了则消费
     */
    @Scheduled(cron = "*/1 * * * * *")
    public void consumer() throws JsonProcessingException {
        List<Message> msgList = delayingQueueService.pull();
        if (null != msgList) {
            long current = System.currentTimeMillis();
            msgList.stream().forEach(msg -> {
                // 已超时的消息拿出来消费
                if (current >= msg.getDelayTime()) {
                    try {
                        log.info("消费消息:{}:消息创建时间:{},消费时间:{}", mapper.writeValueAsString(msg), msg.getCreateTime(), LocalDateTime.now());
                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }
                    //移除消息
                    try {
                        delayingQueueService.remove(msg);
                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }
}

controller中的代码(或自写一个test类):

@RestController
public class RedisMQController { 
    @Resource
    private MessageProvider messageProvider; 

    @PostMapping("/delay/produce")
    public String produce() { 
        //延迟20秒
        messageProvider.sendMessage("同时发送消息1", 20);
        messageProvider.sendMessage("同时发送消息2", 20);
        return "ok";
    }
}

启动项目,访问:http://localhost:8080/delay/produce

后台消费者消费消息:

从输出日志中,可以看出,已经实现了延迟的功能。

自此,Redis实现延迟队列的功能就完成了。

实现延迟队列的其他方案

「RabbitMQ」 :利用 RabbitMQ 做延时队列是比较常见的一种方式,而实际上RabbitMQ自身并没有直接支持提供延迟队列功能,而是通过 RabbitMQ 消息队列的 TTL和 DXL这两个属性间接实现的。

「RocketMQ」 :RocketMQ 发送延时消息时先把消息按照延迟时间段发送到指定的队列中(rocketmq把每种延迟时间段的消息都存放到同一个队列中),然后通过一个定时器进行轮训这些队列,查看消息是否到期,如果到期就把这个消息发送到指定topic的队列中。

「Kafka」 :Kafka支持延时生产、延时拉取、延时删除等,其基于时间轮和 JDK 的 DelayQueue 实现 。

「ActiveMQ」 :需要延迟的消息会先存储在JobStore中,通过异步线程任务JobScheduler将到达投递时间的消息投递到相应队列上 。

 

消息队列对比

总结

如果项目中仅仅是使用个别不是很重要的业务功能,可以使用Redis来做消息队列。但如果对消息可靠性有高度要求的话 ,建议从上面的其他方案中选一个相对合适的来实现。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/395063.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

问一下ChatGPT:DIKW金字塔模型

经常看到这张DIKW金字塔模型图&#xff0c;还看到感觉有点过份解读的图&#xff0c;后面又加上了insight&#xff0c;impact等内容。 Data&#xff1a;是数据&#xff0c;零散的、无规则的呈现到人们眼前&#xff0c;如果你只看到这些数字&#xff0c;如果没有强大的知识背景&a…

STM32之DMA

DMA介绍DMA(Direct MemoryAccess&#xff0c;直接存储器访问)提供在外设与内存、存储器和存储器、外设与外设之间的高速数据传输使用。它允许不同速度的硬件装置来沟通&#xff0c;而不需要依赖于CPU&#xff0c;在这个时间中&#xff0c;CPU对于内存的工作来说就无法使用。DMA…

音乐、音效素材库,好听的BGM都在这~

推荐6个超好用的音频素材网站&#xff0c;免费可商用&#xff0c;热门BGM配乐都能找到&#xff0c;自媒体视频剪辑必备&#xff01;建议收藏&#xff01; 1、菜鸟图库 https://www.sucai999.com/audio.html?vNTYxMjky 菜鸟图库素材非常多&#xff0c;包含了设计、办公、自媒体…

详解FPGA:人工智能时代的驱动引擎观后感

详解FPGA&#xff1a;人工智能时代的驱动引擎观后感 本书大目录 第一章 延续摩尔定律 第二章 拥抱大数据的洪流 第三章 FPGA在人工智能时代的独特优势 第四章 更简单也更复杂——FPGA开发的新方法 第五章 站在巨人肩上——FPGA发展新趋势 文章目录详解FPGA&#xff1a;人工智能…

Redis技术详解

Redis技术详解 Redis是一种支持key-value等多种数据结构的存储系统。可用于缓存&#xff0c;事件发布或订阅&#xff0c;高速队列等场景。支持网络&#xff0c;提供字符串&#xff0c;哈希&#xff0c;列表&#xff0c;队列&#xff0c;集合结构直接存取&#xff0c;基于内存&…

75. CSV文件的写入(保姆级教程)

75. CSV文件的写入&#xff08;保姆级教程&#xff09; 文章目录75. CSV文件的写入&#xff08;保姆级教程&#xff09;1. 目标任务2. 什么是CSV文件2.1 CSV文件知识2.2 准备工作2.3 实操练习12.4 实操练习22.5 实操练习33. os模块文件操作3.1 准备工作3.2 os模块知识回顾3.3 模…

12N60-ASEMI高压MOS管12N60

编辑-Z 12N60在TO-220封装里的静态漏极源导通电阻&#xff08;RDS(ON)&#xff09;为0.7Ω&#xff0c;是一款N沟道高压MOS管。12N60的最大脉冲正向电流ISM为48A&#xff0c;零栅极电压漏极电流(IDSS)为1uA&#xff0c;其工作时耐温度范围为-55~150摄氏度。12N60功耗&#xff…

SpringCloud微服务实战——搭建企业级开发框架(五十):集成移动端推送功能的系统通知公告数据库设计

系统的通知公告功能似乎是很容易被忽略的功能模块&#xff0c;在传统的软件系统中&#xff0c;一般OA类软件系统不可或缺&#xff0c;而在应用软件系统中此功能或有或无&#xff0c;在现在大多数的互联网软件系统中&#xff0c;此功能又必不可缺。所以&#xff0c;在框架设计时…

C++基础了解-04-C++ 变量作用域

变量作用域 一、C 变量作用域 作用域是程序的一个区域&#xff0c;一般来说有三个地方可以定义变量&#xff1a; 1、在函数或一个代码块内部声明的变量&#xff0c;称为局部变量。 2、在函数参数的定义中声明的变量&#xff0c;称为形式参数。 3、在所有函数外部声明的变量…

神经网络结构常见可视化工具汇总及效果演示

文章目录各种训练框架自带的可视化工具pytorch自带pytorchvizkeras自带visualkerasTensorFlow自带TensorBoard其它画图工具NN-SVG&#xff1a;FCNN style、LeNet style、AlexNet stylePlotNeuralNet&#xff1a;使用Latex编写Netron&#xff1a;多种格式的网络模型文件可视化ZE…

计算机网络的166个概念你知道几个 第九部分

计算机网络网络层 路由选择算法&#xff1a;网络层中决定分组发送路径的一种算法。 转发&#xff1a;它指的是将分组从一个输入链路转移到合适的输出链路的动作。 路由选择&#xff1a;指确定分组从一端发送到另一端所选择路径的处理过程。 三种路由交换技术&#xff1a;内…

对word文档中指定的相同内容/文字加颜色或背景颜色

1.请看一段文章。 2.我们要给文中的“code”换成白色字体&#xff0c;黑色背景如何批量呢&#xff1f; 2.1.先摁下“ctrl”"H" 2.2.选择“查找” 2.3.查找内容框填入“code” 2.4.点击以下项中查找&#xff0c;选择主文档。&#xff08;如果你要对某一段文章的相…

Docker操作指南

Docker操作指南 概念解释&#xff1a; 镜像&#xff08;Image&#xff09;&#xff1a;Docker将应用程序及其所需的依赖、函数库、环境、配置等文件打包在一起供运行启动&#xff0c;称为镜像。 容器&#xff08;Container&#xff09;&#xff1a;镜像中的应用程序运行后形…

3.7 最大异或对

异或 二进制位同为0&#xff0c;异为1 异或符号 ^ 异或性质&#xff1a; a^a0 a^00 (a^ b ^c) (a^c ^b) 一道异或的题目 最大异或对 题目链接 思路 注重思维方式 首先是暴力想法&#xff0c;使用两重循环&#xff0c;对每两个数字进行取异或运算&#xff0c;得出最大值考虑…

wireshark 着色规则的含义

文章目录前言wireshark 着色规则的含义前言 如果您觉得有用的话&#xff0c;记得给博主点个赞&#xff0c;评论&#xff0c;收藏一键三连啊&#xff0c;写作不易啊^ _ ^。   而且听说点赞的人每天的运气都不会太差&#xff0c;实在白嫖的话&#xff0c;那欢迎常来啊!!! wires…

【FATE联邦学习debug】 No module named ‘federatedml‘

直接pip install federatedml是无法找得到这个库的。 这个的原因是环境变量的事情&#xff0c;因为在部署文档中&#xff0c;本身提示我们要更新一些环境变量&#xff0c;如果不export那些变量&#xff0c;下面的fate_test其实也是无法测试成功的。 打开bin/init_env.sh&#x…

活动回顾|龙蜥社区标准化SIG首届MeetUp圆满落幕 《龙蜥社区治理规范》征求意见稿正式发布

2 月 25 日周六&#xff0c;龙蜥社区标准化 SIG 召开首次 MeetUp&#xff0c;会上发布了《龙蜥社区治理规范》征求意见稿&#xff0c;致力于通过标准规范开源社区发展&#xff0c;进而推动操作系统行业的规范发展。&#xff08;图/现场合照&#xff09;本次会议共有来自电子标准…

Espressif-IDE v2.8.0 新增功能及开发方向

在乐鑫最近发布的 Espressif-IDE 2.8.0 版本中&#xff0c;我们推出了分区表编辑器和 NVS 分区编辑器功能&#xff0c;优化现有调试器的配置功能并修复多项 Bug &#xff0c;进一步为用户提升了插件质量以及稳定性。 用户可以点此获取最新版本。 • 若您的设备为 Windows 系统…

Python 插件虚拟环境支持

eKuiper 团队于年前发布了 1.8.0 版本&#xff0c;并在 2 月对该版本的文档进行了完善&#xff0c;同时通过 EMQ Demo Day 直播帮助用户更进一步了解新功能的使用场景。 我们也开始了下一个版本 1.9.0 的开发&#xff0c;该版本将是一个较小的迭代版本&#xff0c;主要目标是实…

矩阵打印问题

文章目录1、题目一&#xff1a;正方形矩阵顺时针转动90度1.1 题目描述1.2 思路分析1.3 代码实现2、题目二&#xff1a;转圈打印长方形矩阵2.1 题目描述2.2 思路分析2.3 代码实现3、题目三&#xff1a;zigzag打印矩阵3.1 题目描述3.2 思路分析3.3 代码实现4、题目四&#xff1a;…