Redis队列在项目中实践

Bertha 。 2024-03-26 12:19 79阅读 0赞

##  ##

--------------------

**目录**

系列文章目录

前言

一、Redis 列表(List)

生产者使用 LPUSH 发布消息：

消费者使用 RPOP 拉取消息

二、使用步骤

1.使用rightPop

2.阻塞式拉取消息

leftPop

rightPop

三、 redis 队列的使用场景

机票的航线抓取

火车票的抓取

小说章节更新抓取

总结

--------------------

### 前言 ###

因为Redis List 底层的实现就是一个「链表」，在头部和尾部操作元素，[时间复杂度][Link 1]都是 O(1)，这意味着它非常符合消息队列的模型。

--------------------

### 一、Redis 列表(List) ###

Redis列表是简单的字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头部（左边）或者尾部（右边）

一个列表最多可以包含 232 - 1 个元素 (4294967295, 每个列表超过40亿个元素)。

##### 生产者使用 LPUSH 发布消息： #####

redis 127.0.0.1:6379> lpush queue redis
    1
    redis 127.0.0.1:6379> lpush queue mysql
    2
    redis 127.0.0.1:6379> lpush queue java
    3
    redis 127.0.0.1:6379> lpush queue nginx
    4
    redis 127.0.0.1:6379> LRANGE queue 0 2
    1)  "nginx"
    2)  "java"
    3)  "mysql"

![229f217cf64648648d6964fc72e72f32.jpeg][]

##### 消费者使用 RPOP 拉取消息 #####

redis 127.0.0.1:6379> RPOP queue
    redis
    redis 127.0.0.1:6379> RPOP queue
    mysql
    redis 127.0.0.1:6379> RPOP queue
    java

这个模型非常简单，也很容易理解。

![4c0504209a3f4f8d9df9ddac5324aecf.jpeg][]

### 二、使用步骤 ###

### 1.使用rightPop ###

> 而我们在编写消费者逻辑时，一般是一个「死循环」，这个逻辑需要不断地从队列中拉取消息进行处理

while (true) {
                try {
                    String message = this.stringRedisTemplate.opsForList().rightPop(key);
                    log.info("onMessage.message:{}", message);
    
                    if (StringUtils.isBlank(message)) {
                        continue;
                    }
    
                    handler.handler(message);
                } catch (Exception ex) {
                    log.error("onMessage error !", ex);
                }
            }

如果此时队列为空，那消费者依旧会频繁拉取消息，这会造成「CPU 空转」，不仅浪费 CPU 资源，还会对 Redis 造成压力。

怎么解决这个问题呢？

### 2.阻塞式拉取消息 ###

如果队列为空，消费者在拉取消息时就「阻塞等待」，一旦有新消息过来，就通知我的消费者立即处理新消息呢？

Redis 提供了「阻塞式」拉取消息的命令：BRPOP / BLPOP，这里的 B 指的是阻塞（Block）

![34b1edb46fa2454880f590277e79bdae.jpeg][]

#### **leftPop** ####

有两个重载的方法：

V leftPop(K key);

V leftPop(K key, long timeout, TimeUnit unit);

此方法会弹出list头部的元素（弹出后就在list中删除），跟栈一样，重载的那个方法，如果list不存在数据，则会阻塞住，等有数据了就会返回（最长阻塞时间就是设置的超时时间）。该方法对应到redis的操作是BLPOP（blocking left pop）

#### **rightPop** ####

与leftPop类似，只是从list尾部弹出数据

代码如下（示例）：

private void onMessage() {
            while (true) {
                try {
                    /**
                     * 此方法会弹出list头部的元素（弹出后就在list中删除），跟栈一样，重载的那个方法，如果list不存在数据，则会阻塞住，
                     * 等有数据了就会返回（最长阻塞时间就是设置的超时时间）。
                     * 该方法对应到redis的操作是BLPOP（blocking left pop）
                     */
                    String message = stringRedisTemplate.opsForList().rightPop(spiderCacheEnum.getKey(), 10, TimeUnit.SECONDS);
                    log.info("onMessage.message:{}", message);
    
                    if (StringUtils.isBlank(message)) {
                        continue;
                    }
    
    
                    handler.handler(message);
                } catch (Exception ex) {
                    log.error("onMessage error !", ex);
                }
            }
        }

--------------------

### 三、 redis 队列的使用场景 ###

在程序开发中，消息队列常见的使用场景其实有很多: 系统解耦 异步处理 流量削峰 日志处理 。但有一些业务场景，使用kafka ，rocketmq 等消息队列又显的系统过重。比如以下场景可以使用轻量级的 redis 队列 解决。

##### 机票的航线抓取 #####

近实时的数据抓取\[机票的剩余票数\]，我们需要近实时刷新机票的剩余票数，需要不间断的访问航信接口查询航班信息和机票信息，比如3分钟抓取一次热门航线，10分钟抓取非热门航线。需要一个生产者每三分钟生成一个 \{"dep":"北京", "dest":"上海","departure":"2023-06-26"\} 的任务。消费者根据这个任务抓取数据。设计如下：

![a88dadb0f3c64109ae1303f52fd1219b.jpeg][]

##### 火车票的抓取 #####

近实时的数据抓取\[车票的剩余票数\]，可以使用定时任务，比如10分钟抓取一次，如果抓取失败，不会影响业务逻辑，后续还会再次抓取该车次数据。类似机票抓取逻辑，不再赘述。

##### 小说章节更新抓取 #####

一些小说阅读网站，小说的章节更新也是采用类似方案。我们需要更新小说的章节内容，但又不知道小说源的更新时间，所以需要不间断的check 小说源是否有章节更新，如果有章节更新则同步更新小说信息，如果没有章节更新则间断10分钟再次check 小说源是否有章节更新。

### 四、 代码演示 ###

##### 队列接口定义 #####

/**
     * 阻塞传队列接口
     *
     * @author yangyanping
     * @date 2023-06-26
     */
    public interface Queue<T> {
    
        /**
         * 用来向队尾存入元素
         */
        void push(T t);
    
        /**
         * 用来从队首取元素。如果队列空，则会等待 timeout 时间后进行一次重试，
         * 如果取不到，则返回null；否则返回取得的元素
         */
        T poll(long timeout, TimeUnit unit);
    }

##### 小说抓取队列任务 #####

/**
     * 小说塞传队列
     *
     * @author yangyanping
     * @date 2023-06-26
     */
    @Slf4j
    @Component
    public class BookGrabQueueImpl implements Queue<BookGrabTask> {
        private static final String TASK_KEY = "book.grab.task.queue";
    
        @Resource(name = "stringCacheTemplate")
        private CacheService<String, String> cacheService;
    
        /**
         * 用来向队尾存入元素
         */
        @Override
        public void push(BookGrabTask task) {
            cacheService.leftPush(TASK_KEY, JSON.toJSONString(task));
        }
    
        /**
         * 用来从队首取元素。如果队列空，则会等待 timeout 时间后进行一次重试，
         * 如果取不到，则返回null；否则返回取得的元素
         */
        @Override
        public BookGrabTask poll(long timeout, TimeUnit unit) {
            String json = cacheService.rightPop(TASK_KEY, timeout, unit);
    
            if (StringUtils.isBlank(json)) {
                return null;
            }
    
    
            return JSON.parseObject(json, BookGrabTask.class);
        }
    }

##### 书籍任务生成定时器 #####

/**
     * 书籍抓取任务入队--定时器
     *
     * @author yangyanping
     * @date 2022-06-23
     */
    @Slf4j
    @Component
    public class BookGrabTaskProduceScheduler {
    
        @Resource(name = "bookGrabQueueImpl")
        private Queue bookGrabQueueImpl;
    
         /**
         * 生产任务
         */
        @XxlJob("bookGrabTaskProduceScheduler")
        public ReturnT<String> execute(String param) {
    
            BookGrabTask task = new BookGrabTask();
            task.setBookId(1);
            //doBuiness()
            bookGrabQueueImpl.push(task);
    
            return ReturnT.SUCCESS;
        }
    }

书籍任务消费代码

/**
     * 书籍抓取分布式任务消费
     *
     * @author yangyanping
     * @date 2023-06-25
     */
    @Slf4j
    @Component
    public class BookGrabConsumer implements ApplicationRunner {
    
        private final ExecutorService threadPool = new ThreadPoolExecutor(5, 5,
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(5), new ThreadPoolExecutor.CallerRunsPolicy());
    
    private static final String distributed_Key = "book.grab.id:%s";
    
        @Resource(name = "bookGrabQueueImpl")
        private Queue bookGrabQueueImpl;
    
        @Override
        public void run(ApplicationArguments args) throws Exception {
            log.info("BookGrabConsumer...... is running");
    
             threadPool.execute(() -> this.execute());
        }
    
         /**
         * 执行回传任务
         */
        private void execute() {
            while (true) {
                boolean acquireLock = false;
                String reqId = UUID.fastUUID().toString();
                String distributedKey = "";
    
                try {
                    BookGrabTask task = bookGrabQueueImpl.poll(5, TimeUnit.SECONDS);
    
                    if (task == null) {
                        continue;
                    }
    
                    //分布式锁key
                    distributedKey = String.format(distributed_Key, task.getBookId());
                    //申请分布式锁
                    acquireLock = DistributedUtil.acquire(cacheService, distributedKey, reqId, 20, TimeUnit.MINUTES);
    
                    if (acquireLock) {
                        // doBuiness();
                    }
                } catch (Exception ex) {
                    log.error("BookGrabConsumer.execute fail", ex);
                } finally {
                    if (acquireLock) {
                        DistributedUtil.release(cacheService, distributedKey, reqId);
                    }
                }
            }
        }
    }

##### 补充知识 ApplicationRunner 和 CommandLineRunner #####

Spring Boot 应用，在启动的时候，如果想做一些事情，比如预先加载并缓存某些数据，读取某些配置等等。**总而言之，做一些初始化的操作时，那么 Spring Boot 就提供了两个接口帮助我们实现。**

这两个接口是：

*  **ApplicationRunner 接口**
 *  **CommandLineRunner 接口**

这两个接口可以在 Spring 的环境下指定一个 Bean 运行（run）某些你想要做的事情，如果你有多个 Bean 进行指定，那么可以通过 Ordered 接口或者 @Order 注解指定执行顺序。

## 总结 ##

这种队列模型，有什么缺点？

我们一起来分析一下：

1.  不支持重复消费：消费者拉取消息后，这条消息就从 List 中删除了，无法被其它消费者再次消费，即不支持多个消费者消费同一批数据
2.  消息丢失：消费者拉取到消息后，如果发生异常宕机，那这条消息就丢失了

[Link 1]: https://so.csdn.net/so/search?q=%E6%97%B6%E9%97%B4%E5%A4%8D%E6%9D%82%E5%BA%A6&spm=1001.2101.3001.7020
[229f217cf64648648d6964fc72e72f32.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/350b1a3e63084a95ba5d888b4cbf7c2f.png
[4c0504209a3f4f8d9df9ddac5324aecf.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/5b12b0a3302242f690254c716d9fc1da.png
[34b1edb46fa2454880f590277e79bdae.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/4117ef6d79b0463085d83c2032db0b32.png
[a88dadb0f3c64109ae1303f52fd1219b.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/71679e099f1c44ab9cbe8740d03b3adb.png