正文

redis之消息队列实现

author  author  2023-03-05  420

关键词：

文章目录

秒杀场景

采用消息队列实现

List实现消息队列
PubSub（发布订阅）实现消息队列
基于Stream实现消息队列

消费者组

实践

总结

秒杀问题是非常重要且比较难实现的，如果不进行架构的优化的话，直接访问会给业务系统造成很大的压力…

秒杀场景

场景一：双十一出现的秒杀场景，比如部分商品开展活动，特价11.11，但是库存一般不多，甚至只有一台。在并发量比较大的情况下，如果我们不进行一些优化的话，很容易出现线程安全问题，而且可能给系统带来太大压力，导致宕机。

如果采用以下的秒杀架构图：

Redis之消息队列实现_List

我们可以采用Redis来保证一人一单，我们可以将判断秒杀库存和校验一人一单放到Redis层面来完成，采用异步下单来进行秒杀。

这里我们就需要使用到全面分布式锁中生成的全局ID生成器，因为订单id是需要满足自增、唯一性的条件的，这里也可以采用一些网上的开源方案雪花算法等衍生都可以，但是这样还是会存在问题。什么问题呢？

下单操作的原子性？同步队列的实现？

下单代码是需要校验一人一单和库存等条件的，需要使用lua脚本来保证操作的原子性。
采用阻塞队列来存放数据存在内存限制问题和数据安全问题

采用消息队列实现

消息队列（Message Queue)，也叫存放消息的队列，最简单的消息队列模型包括：

消息队列：存储和管理消息，也叫做消息代理。
生产者：发送消息到消息队列
消费者：从消息队列获取消息并处理消息

$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mtsozweI-1677297911799)(C:\\Users\\DY\\AppData\\Roaming\\marktext\\images\\2023-02-21-10-17-22-image.png)]$

在Redis中也提供了三种方式实现消息队列：

List结构：基于List模拟消息队列
PubSub：点对点的消息模型
Stream：比较完善的消息队列模型

List实现消息队列

在Redis中List的数据结构本身就是一个双向链表，所以很容易就可以模拟出队列的效果。

我们可以通过LPUSH结合RPOP、RPUSH结合LPOP实现，不过当队列中没有消息时RPOP或LPOP操作会返回null，并不会像阻塞队列一样等待消息，因此我们也可以使用BRPOP或者BLPOP来实现阻塞的效果。

优点

优点非常明显，不再受限于JVM内存（阻塞队列消费）
采用Redis的持久化机制，数据安全有保证
可以保证消息的有序性

缺点：

无法避免消息丢失
仅支持单消费

PubSub（发布订阅）实现消息队列

消费者可以订阅一个或多个channel，生产者发送对应channel消息后，所以订阅者都能收到相关消息。

乍一看，好像这个确实比List强多了，使用发布订阅模式，还支持多订阅。

$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-JJtSFlh0-1677297911800)(C:\\Users\\DY\\AppData\\Roaming\\marktext\\images\\2023-02-21-10-37-09-image.png)]$

但是缺点也很明显：

不支持数据持久化
无法避免消息丢失
消息堆积有限制，超过时也会出现数据丢失

基于Stream实现消息队列

Stream是Redis5.0以后提出的一种数据类型，是功能比较完善的消息队列

发送消息：

$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fWu2DdL1-1677297911801)(C:\\Users\\DY\\AppData\\Roaming\\marktext\\images\\2023-02-21-10-44-29-image.png)]$

读取消息：

$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KJKu2Q53-1677297911802)(C:\\Users\\DY\\AppData\\Roaming\\marktext\\images\\2023-02-21-10-44-50-image.png)]$

除去可以实现读取外，还可以设置阻塞时间，从而实现持续监听的效果

$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-V62KOP2a-1677297911802)(C:\\Users\\DY\\AppData\\Roaming\\marktext\\images\\2023-02-21-10-47-02-image.png)]$

上面的ID指定为$时，标识读取的是最新的消息，如果有同时两条消息同时到达MQ，下次读取可能会出现漏读的问题

消费者组

将多个消费者划分到一个组中，监听同一个队列。

$[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-MtsRDJTc-1677297911803)(C:\\Users\\DY\\AppData\\Roaming\\marktext\\images\\2023-02-21-11-32-58-image.png)]$

在Redis的Stream的ID策略中

‘>’: 从下一条未消费的消息开始
其它：根据指定id从pending-list中获取已消费但是未确认的消息

这里的确认是因为前面提到可能会丢失消息，而Stream中的消息在经过消费后，需要进行XACK确认，如果没有进行确认就会把这条消息放到pending-list中。

实践

前面提到如何保证秒杀下单的原子性需要使用到lua脚本，在这里面我把对应的key都写死了

-- 1.参数列表
-- 1.1.优惠券id
local voucherId = ARGV[1]
-- 1.2.用户id
local userId = ARGV[2]
-- 1.3.订单id
local orderId = ARGV[3]

-- 2.数据key
-- 2.1.库存key
local stockKey = seckill:stock: .. voucherId
-- 2.2.订单key
local orderKey = seckill:order: .. voucherId

-- 3.脚本业务
-- 3.1.判断库存是否充足 get stockKey
if(tonumber(redis.call(get, stockKey)) <= 0) then
    -- 3.2.库存不足，返回1
    return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call(sismember, orderKey, userId) == 1) then
    -- 3.3.存在，说明是重复下单，返回2
    return 2
end
-- 3.4.扣库存 incrby stockKey -1
redis.call(incrby, stockKey, -1)
-- 3.5.下单（保存用户）sadd orderKey userId
redis.call(sadd, orderKey, userId)
-- 3.6.发送消息到队列中， XADD stream.orders * k1 v1 k2 v2 ...
redis.call(xadd, stream.orders, *, userId, userId, voucherId, voucherId, id, orderId)
return 0

在消费中，采用ExectorService来创建一个线程池来进行订单的消费

private class VoucherOrderHandler implements Runnable 
        String queueName = "stream.orders";
        @Override
        public void run() 
            while (true) 
                try 
                    // 1. 获取消息队列中的信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.order >
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(queueName, ReadOffset.lastConsumed())
                    );
                    // 2. 判断消息获取是否成功
                    if (list == null || list.isEmpty()) 
                        // 2.1 获取失败，没有消息，进行下一次循环
                        continue;
                    
                    // 3. 解析消息中的订单信息
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> values = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
                    // 3. 获取成功，可以下单
                    handleVoucherOrder(voucherOrder);
                    // 4. ACK确认 XACK stream.orders  g1 id
                    stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
                 catch (Exception e) 
                    log.error("处理订单异常");
                    handlePendingLists();
                
            
        

        private void handlePendingLists() 
            while (true) 
                try 
                    // 1. 获取pendingList队列中的信息 XREADGROUP GROUP g1 c1 COUNT 1 STREAMS stream.order 0
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create(queueName, ReadOffset.from("0"))
                    );
                    // 2. 判断消息获取是否成功
                    if (list == null || list.isEmpty()) 
                        // 2.1 获取失败，说明pending-list没有消息，结束循环
                        break;
                    
                    // 3. 解析消息中的订单信息
                    MapRecord<String, Object, Object> record = list.get(0);
                    Map<Object, Object> values = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
                    // 3. 获取成功，可以下单
                    handleVoucherOrder(voucherOrder);
                    // 4. ACK确认 XACK stream.orders  g1 id
                    stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
                 catch (Exception e) 
                    log.error("处理pend-list订单异常", e);
                    try 
                        Thread.sleep(20);
                     catch (InterruptedException ex) 
                        ex.printStackTrace();

总结

Redis只能满足小项目的需求，更大需求可能需要用到更加高级的消息队列，比如Kafka、RabbitMQ、RocketMQ等等，还有好多学问要学习呢…

java之springboot集成redis实现消息队列(代码片段)

Java之SpringBoot集成redis实现消息队列一、设置好Redis的配置文件（application.yml）二、消息接收者实体类（RedisMessage）三、消息队列订阅者（RedisSubConfig）四、发送消息（消息生产者）一、设置好Redis的... 查看详情

redis之消息队列实现

文章目录秒杀场景采用消息队列实现List实现消息队列PubSub（发布订阅）实现消息队列基于Stream实现消息队列消费者组实践总结秒杀问题是非常重要且比较... 查看详情

什么鬼，面试官竟然让我用redis实现一个消息队列！！？(代码片段)

什么鬼，面试官竟然让我用Redis实现一个消息队列！！？△Hollis,一个对Coding有着独特追求的人△这是Hollis的第241篇原创分享作者l小胖儿来源lHollis（ID：hollischuang）众所周知，redis是一个高性能的分布式key-value存储系统，在NoSQL数... 查看详情

elkstack之消息队列

redis消息队列安装redisyum-yinstallredis修改配置文件修改ip后台运行启动systemctlstartredis查看lsof-i:6379连接redis-cli-h10.13.85.9 cd/etc/logstash/conf.d/vimredis.confinput{stdin{}}output{redis{ host=>"10.13.85.9" port 查看详情

redis数据类型之—list

（1）list简单介绍list是一个有序的字符串列表，是使用双向列表实现的，可以实现最新消息排行、消息队列等功能。（2）list常用命令查看详情

一个redis消息队列实现

消息队列某次在某乎上看到有人提到消息队列的问题，然后有人在回答里提到了Redis，接着便有人在评论里指出：Redis是缓存，不是消息队列。但不幸的是，Redis的确提供一个简易的消息队列机制，可以用于... 查看详情

一个redis消息队列实现

redis基于(listpubsubstream消费者组)实现消息队列，基于stream结构实现异步秒杀下单(代码片段)

(目录)上一篇博文部分：Redis消息队列1、Redis消息队列-认识消息队列什么是消息队列？最简单的消息队列模型包括3个角色：使用队列的好处在于解耦：这种场景在秒杀中就变成了：这里我们可以使用一些现成的mq，比如kafka，rabbi... 查看详情

javaweb项目架构之redis分布式日志队列

...实就是一个日志收集的功能，只不过中间加了一个Redis做消息队列罢了。前言为什么需要消息队列？当系统中出现“生产“和“消费“的速度或稳定性等因素不一致的时候，就需要消息队列，作为抽象层，弥合双方的差异。比如... 查看详情

redis消息队列有没有

参考技术A基于Redis消息队列-实现短信服务化1.Redis实现消息队列原理常用的消息队列有RabbitMQ,ActiveMQ，个人觉得这种消息队列太大太重，本文介绍下基于Redis的轻量级消息队列服务。一般来说，消息队列有两种模式，一种是发布者... 查看详情

redis的基本使用（二）消息队列

参考技术A使用消息中间件的时候，并非每次都需要非常专业的消息中间件，假如只有一个消息队列，只有一个消费者，那就没有必要去使用专业的消息中间件，这种情况可以直接使用Redis来做消息队列。Redis的消息队列不是特别... 查看详情

使用redis的发布订阅模式实现消息队列

配置文件<?xmlversion="1.0"encoding="UTF-8"?><beansxmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:aop="http://www.springframework.org 查看详情

消息队列之利器锋芒(代码片段)

...所用的系统越来越复杂。系统势必会发展成分布式系统。消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋，可靠投递，广播，最终一致性等问题。实现高性能，高可用，可伸缩和最终一致性... 查看详情

在springboot中用redis实现消息队列

环境依赖创建一个新的springboot工程，在其pom文件,加入spring-boot-starter-data-redis依赖：<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifact 查看详情

redis使用list实现消息队列的利与弊(代码片段)

分布式系统中必备的一个中间件就是消息队列，通过消息队列我们能对服务间进行异步解耦、流量消峰、实现最终一致性。目前市面上已经有RabbitMQ、RochetMQ、ActiveMQ、Kafka等，有人会问：“Redis适合做消息队列么？... 查看详情

javaweb项目架构之redis分布式日志队列-springboot实例

...实就是一个日志收集的功能，只不过中间加了一个Redis做消息队列罢了。? 为什么需要消息队列？当系统中出现“生产“和“消费“的速度或稳定性等因素不一致的时候，就需要消息队列，作为抽象层，弥合双方... 查看详情

一个redis消息队列实现

消息队列为啥用redis实现

...信息(请求内容，请求方标识等等).参考技术A因为redis实现消息队列很简单！$this->redis->rPush($key, $val); // 右边入$this->redis->lPop($key); // 左边出本回答被提问者采纳查看详情