文章详情

登录

短信预约-IT技能 免费直播动态提醒

选择考试省份

  • 北京
  • 上海
  • 天津
  • 重庆
  • 河北
  • 山东
  • 辽宁
  • 黑龙江
  • 吉林
  • 甘肃
  • 青海
  • 河南
  • 江苏
  • 湖北
  • 湖南
  • 江西
  • 浙江
  • 广东
  • 云南
  • 福建
  • 海南
  • 山西
  • 四川
  • 陕西
  • 贵州
  • 安徽
  • 广西
  • 内蒙
  • 西藏
  • 新疆
  • 宁夏
  • 兵团

请输入下面的图形验证码

提交验证

短信预约提醒成功

基于Redis实现消息队列的实践

2024-11-30 02:51

关注

先来回顾一个Redis有哪些特性：

  1. 速度快：Redis是基于内存的key-value类型的数据库，数据都存放在内存中，使得读写速度非常快，能够达到每秒数十万次的读写操作。
  2. 键值对的数据结构：Redis中的数据以键值对的形式存储，使得查询和操作数据非常方便和高效。
  3. 功能丰富：Redis具有许多实用的功能，例如键过期、发布订阅、Lua脚本、事务和管道等。这些功能使得Redis能够广泛应用于各种场景，如缓存、消息系统等。
  4. 持久化：Redis提供了两种持久化方案，即RDB（根据时间生成数据快照）和AOF（以追加方式记录每次写操作）。两种方案可以互相配合，确保数据的安全性。
  5. 主从复制：Redis支持主从复制功能，可以轻松实现数据备份和扩展。主节点会将其数据复制给从节点，从而实现数据的冗余和备份。
  6. 高可用和分布式：Redis从2.8版本开始提供了高可用实现哨兵模式，可以保证节点的故障发现和故障自动转移。此外，Redis从3.0版本开始支持集群模式，可以轻松实现数据的分布式存储和扩展。

总结一下：redis的特点就是：快、简单、稳定；

以RocketMQ为代表，作为专业的消息中间件而言，有哪些特性呢：

  1. 高性能、高可靠：RocketMQ采用分布式架构，能够高效地处理大量消息，同时也具有高可靠性的特性，能够保证消息的不丢失和正确传递。
  2. 高实时：RocketMQ支持消息的实时传递，能够满足实时交易系统的需求，为系统提供及时、准确的消息。
  3. 事务消息：RocketMQ支持事务消息，能够在消息发送和接收过程中保持事务的一致性，确保消息的可靠性和系统的稳定性。
  4. 顺序消息：RocketMQ可以保证消息的有序性，无论是在一个生产者还是多个生产者之间，都能保证消息按照发送顺序进行消费。
  5. 批量消息：RocketMQ支持批量消息，能够一次性发送多条消息，提高消息发送效率。
  6. 定时消息：RocketMQ支持定时消息，能够在指定的时间将消息发送到指定的Topic，满足定时任务的需求。
  7. 消息回溯：RocketMQ支持消息回溯，能够根据需要将消息重新发送到指定的Topic，便于调试和错误处理。
  8. 多种消息模式：RocketMQ支持发布/订阅、点对点、群聊等多种消息模式，适用于不同的业务场景。
  9. 可扩展性：RocketMQ采用分布式架构，能够方便地扩展消息处理能力，支持多个生产者和消费者同时处理消息。
  10. 多语言支持：RocketMQ提供多种语言的客户端库，支持包括Java、Python、C++等在内的多种编程语言。

总结一下：RocketMQ的特点就是除了性能非常高、系统本身的功能比较专业、完善，能适应非常多的场景；

从上述分析可以看出，Redis队列和MQ消息队列各有优势，Redis的最大特点就是快，所以基于Redis的消息队列相比MQ消息队列而言，更适合实时处理，但是基于Redis的消息队列更易受服务器内存限制；而RocketMQ消息队列作为专业的消息中间件产品，功能更完善，更适合应用于比较复杂的业务场景，可以实现离线消息发送、消息可靠投递以及消息的安全性，但MQ消息队列的读写性能略低于Redis队列。在技术选型时，除了上述的因素外，还有一个需要注意：大多数系统都会引入Redis作为基础的缓存中间件使用，如果要选用RocketMQ的话，还需要额外再申请资源进行部署。

很多时候，所谓的优点和缺点，只是针对特定场景而言，如果场景不一样了，优点可能会变成缺点，缺点也可能会变成优点。因此，除了专业的消息中间件外，基于Redis实现一个消息队列也是有必要的，在某些特殊的业务场景，比如一些并发量不是很高的管理系统，某些业务流程需要异步化处理，这时选择基于Redis自己实现一个消息队列，也是一个比较好的选择。这也是本篇文章主要分享的内容。

消息队列的基础知识：

什么是队列？

队列（Queue）是一种数据结构，遵循先进先出（FIFO）的原则。在队列中，元素被添加到末尾（入队），并从开头移除（出队）。

Java中有哪些队列？

  1. LinkedList：LinkedList实现了Deque接口，可以作为队列（FIFO）或栈（LIFO）使用。它是一个双向链表，所以插入和删除操作具有很高的效率。
  2. ArrayDeque：ArrayDeque也是一个双端队列，具有高效的插入和删除操作。与LinkedList相比，ArrayDeque通常在大多数操作中表现得更快，因为它在内部使用动态数组。
  3. PriorityQueue：PriorityQueue是一个优先队列，它保证队列头部总是最小元素。你可以自定义元素的排序规则。
  4. ConcurrentLinkedQueue：ConcurrentLinkedQueue是一个线程安全的队列，它使用无锁算法进行并发控制。它适用于高并发场景，但在低并发场景中可能比其他队列慢。
  5. LinkedBlockingQueue：LinkedBlockingQueue是一个线程安全的阻塞队列，它使用链表数据结构来存储数据。当队列为空时，获取元素的操作将会被阻塞；当队列已满时，插入元素的操作将会被阻塞。
  6. ArrayBlockingQueue：ArrayBlockingQueue是一个线程安全的阻塞队列，它使用数组数据结构来存储数据。与LinkedBlockingQueue相比，ArrayBlockingQueue的容量是固定的。
  7. PriorityBlockingQueue：PriorityBlockingQueue是一个线程安全的优先阻塞队列。与PriorityQueue类似，它保证队列头部总是最小元素。
  8. SynchronousQueue：SynchronousQueue是一个线程安全的阻塞队列，它只包含一个元素。当队列为空时，获取元素的操作将会被阻塞；当队列已满时，插入元素的操作将会被阻塞。
  9. DelayQueue：DelayQueue是一个无界阻塞队列，用于放置实现了Delayed接口的对象，其中的对象只能在其到期时才能从队列中取走。

LinkedBlockingQueue

以LinkedBlockingQueue为例，其使用方法是这样的：

创建了一个生产者线程和一个消费者线程，生产者线程和消费者线程分别对同一个LinkedBlockingQueue对象进行操作。生产者线程通过调用put()方法将元素添加到队列中，而消费者线程通过调用take()方法从队列中取出元素。这两个方法都会阻塞线程，直到队列中有元素可供取出或有空间可供添加元素。

import java.util.concurrent.LinkedBlockingQueue;  
  
public class LinkedBlockingQueueExample {  
    public static void main(String[] args) {  
        LinkedBlockingQueue queue = new LinkedBlockingQueue<>();  
  
        // 生产者线程  
        new Thread(() -> {  
            for (int i = 0; i < 10; i++) {  
                try {  
                    queue.put("Element " + i);  
                    System.out.println("Produced: Element " + i);  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                }  
            }  
        }).start();  
        // 消费者线程  
        new Thread(() -> {  
            for (int i = 0; i < 10; i++) {  
                try {  
                    String element = queue.take();  
                    System.out.println("Consumed: " + element);  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                }  
            }  
        }).start();  
    }  
}


基于Redis实现消息队列的几种方式

基于List数据类型

图片

基于Zset数据类型

基于发布订阅模式

无法持久化保存消息，如果 Redis 服务器宕机或重启，那么所有的消息将会丢失；

发布订阅模式是“发后既忘”的工作模式，如果有订阅者离线重连之后就不能消费之前的历史消息；

不支持消费者确认机制，稳定性不能得到保证，例如当消费者获取到消息之后，还没来得及执行就宕机了。因为没有消费者确认机制，Redis 就会误以为消费者已经执行了，因此就不会重复发送未被正常消费的消息了，这样整体的 Redis 稳定性就被没有办法得到保障了。

基于Stream类型

基于Stream 类型实现：使用 Stream 的 xadd 和 xrange 来实现消息的存入和读取了，并且 Stream 提供了 xack 手动确认消息消费的命令，用它我们就可以实现消费者确认的功能了，使用命令如下：

127.0.0.1:6379> xack mq group1 1580959593553-0
(integer) 1

消费确认增加了消息的可靠性，一般在业务处理完成之后，需要执行 ack 确认消息已经被消费完成，整个流程的执行如下图所示：

其中“Group”为群组，消费者也就是接收者需要订阅到群组才能正常获取到消息。

以上就是基于Redis实现消息队列的几种方式的简单对比介绍，下面主要是分享一下基于Redis的List数据类型实现，其他几种方式，有兴趣的小伙可以自己尝试一下。

基于Redis的List数据类型实现消费队列的工作原理是什么？

Redis基于List结构实现队列的原理主要依赖于List的push和pop操作。

在Redis中，你可以使用LPUSH命令将一个或多个元素推入列表的左边，也就是列表头部。同样，你可以使用RPUSH命令将一个或多个元素推入列表的右边，也就是列表尾部。

对于队列来说，新元素总是从队列的头部进入，而读取操作总是从队列的尾部开始。因此，当你想将一个新元素加入队列时，你可以使用LPUSH命令。当你想从队列中取出一个元素时，你可以使用RPOP命令。

此外，Redis还提供了BRPOP命令，这是一个阻塞的RPOP版本。如果给定列表内没有任何元素可供弹出的话，将阻塞连接直到等待超时或发现可弹出元素为止。

需要注意的是，虽然Redis能够提供原子性的push和pop操作，但是在并发环境下使用队列时，仍然需要考虑线程安全和并发控制的问题。你可能需要使用Lua脚本或者其他机制来确保并发操作的正确性。

总的来说，Redis通过提供List数据结构以及一系列相关命令，可以很方便地实现队列的功能。

下面是Redis关于List数据结构操作的命令主要包括以下几种：

基于Redis的List数据类型实现延迟消息队列实战

需求描述

以一个实际需求为例，演示一个基于Redis的延迟队列是怎么使用的？

有一个XX任务管理的功能，主要的业务过程：

创建任务后；

不断检查任务的状态，任务的状态有三种：待执行、执行中、执行完成；

如果任务状态是执行完成后，主动获取任务执行结果，对任务执行结果进行处理；如果任务状态是待执行、执行中，则延迟5秒后，再次查询任务执行状态；

图片

实现方案

依赖引入


    org.springframework.boot
    spring-boot-starter-redis
    1.4.7.RELEASE


    org.redisson
    redisson
    3.23.1

定义三个延迟队列BeforeQueue、RunningQueue、CompleteQueue，对队列的任务进行存取，BeforeQueue用于对待执行状态的任务的存取，Running用于对执行中状态的任务的存取，CompleteQueue用于对执行完成状态的任务的存取，在三个任务队列中，取出元素是阻塞的，即如果队列中没有新的任务，当前线程会一直阻塞等待，直到有新的任务进入；如果是队列中还有元素，则遵循先进先出的原则逐个取出进行处理；

@Component
@Slf4j
public class BeforeQueue {
    @Autowired
    private RedissonClient redissonClient;


    
    public Object take(){
        RBlockingQueue queue1 = redissonClient.getBlockingQueue("queue1");
        Object obj = null;
        try {
            obj = queue1.take();
            log.info("从myqueue1取出元素:{}",obj.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return obj;
    }


    
    public void offer(Object obj){
        RBlockingDeque queue1 = redissonClient.getBlockingDeque("queue1");
        RDelayedQueue delayedQueue = redissonClient.getDelayedQueue(queue1);
        delayedQueue.offer(obj,5, TimeUnit.SECONDS);
        log.info("向myqueue1设置元素:{}",obj.toString());
    }
}
@Component
@Slf4j
public class RunningQueue {
    @Autowired
    private RedissonClient redissonClient;


    public Object take(){
        RBlockingQueue queue1 = redissonClient.getBlockingQueue("queue2");
        Object obj = null;
        try {
            obj = queue1.take();
            log.info("从myqueue2取出元素:{}",obj.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return obj;
    }


    public void offer(Object obj){
        RBlockingDeque queue1 = redissonClient.getBlockingDeque("queue2");
        RDelayedQueue delayedQueue = redissonClient.getDelayedQueue(queue1);
        delayedQueue.offer(obj,5, TimeUnit.SECONDS);
        log.info("向myqueue2设置元素:{}",obj.toString());
    }
}
@Component
@Slf4j
public class CompleteQueue {
    @Autowired
    private RedissonClient redissonClient;


    public Object take(){
        RBlockingQueue queue1 = redissonClient.getBlockingQueue("queue3");
        Object obj = null;
        try {
            obj = queue1.take();
            log.info("从CompleteQueue取出元素:{}",obj.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return obj;
    }


    public void offer(Object obj){
        RBlockingQueue queue1 = redissonClient.getBlockingDeque("queue3");
        RDelayedQueue delayedQueue = redissonClient.getDelayedQueue(queue1);
        delayedQueue.offer(obj,5, TimeUnit.SECONDS);
        log.info("向CompleteQueue设置元素:{}",obj.toString());
    }
}
定义三个监听器BeforeQueueListener、RunningQueueListener、CompleteQueueListener，监听器的主要作用主要就是负责监听三个队列中是否有新的任务 元素进入，如果有，则立即取出消费；如果没有，则阻塞等待新的元素进入，具体的实现逻辑是：新创建的任务会先放置到BeforeQueue中，BeforeQueueListener监听到有新的任务进入，会取出任务作一些业务处理，业务处理完一放入到RunningQueue中，RunningQueueListener监听到有新的任务进入，会取出任务再进行处理，这里的处理主要是查询任务执行状态，查询状态结果主要分两种情况：1、执行中、待执行状态，则把任务重新放入RunningQueue队列中，延迟5秒；2、执行完成状态，则把任务放置到CompleteQueue中；CompleteQueueListener监听到有新的任务进入后，会主动获取任务执行结果，作最后业务处理；
监听器在在处理队列中的数据相关的业务时，如果发生异常，则需要把取出的元素再重新入入到当前队列中，等待下一轮的重试；
@Component
@Slf4j
public class BeforeQueueListener implements Listener{
    @Autowired
    private BeforeQueue beforeQueue;
    @Autowired
    private RunningQueue runningQueue;
    @Override
    public void start() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true){
                    log.info("监听器进入阻塞:BeforeQueueListener");
                    Object obj = beforeQueue.take();
                    if (ObjectUtil.isNotNull(obj)) {
                        try {
                            log.info("开始休眠1s模拟业务处理:BeforeQueueListener,元素:{}",obj.toString());
                            Thread.currentThread().sleep(1000);
                            log.info("业务处理完成:BeforeQueueListener,元素:{}",obj.toString());
                            runningQueue.offer(obj);
                        } catch (InterruptedException e) {
                            log.error("业务处理发生异常,重置元素到BeforeQueue队列中");
                            log.error(e.getMessage());
                            beforeQueue.offer(obj);
                        }


                    }
                }
            }
        }).start();
    }
}
@Component
@Slf4j
public class RunningQueueListener implements Listener {


    @Autowired
    private RunningQueue runningQueue;
    @Autowired
    private CompleteQueue completeQueue;


    @Override
    public void start() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    log.info("监听器进入阻塞:RunningQueueListener");
                    Object obj = runningQueue.take();
                    if (ObjectUtil.isNotNull(obj)) {
                        try {
                            log.info("开始休眠1s模拟业务处理:RunningQueueListener,元素:{}", obj.toString());
                            Thread.currentThread().sleep(1000);
                            Random random = new Random();
                            int i = random.nextInt(2);
                            if (i==0) {
                                test();
                            }
                            log.info("业务处理完成:RunningQueueListener,元素:{}", obj.toString());
                            completeQueue.offer(obj);
                        } catch (Exception e) {
                            log.error("业务处理发生异常,重置元素到RunningQueue队列中");
                            log.error(e.getMessage());
                            runningQueue.offer(obj);
                        }
                    }
                }
            }
        }).start();
    }


    public void test(){
        try {
            int i=1/0;
        } catch (Exception e) {
           throw  new RuntimeException("除数异常");
        }
    }


}
@Component
@Slf4j
public class CompleteQueueListener implements Listener{


    @Autowired
    private CompleteQueue completeQueue;
    @Override
    public void start() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true){
                    log.info("监听器进入阻塞:CompleteQueueListener");
                    Object obj = completeQueue.take();
                    if (ObjectUtil.isNotNull(obj)) {
                        try {
                            log.info("开始休眠1s模拟业务处理:CompleteQueueListener,元素:{}",obj.toString());
                            Thread.currentThread().sleep(1000);
                            log.info("业务处理完成:listener3,元素:{}",obj.toString());
                        } catch (InterruptedException e) {
                            log.error("业务处理发生异常,重置元素到CompleteQueue队列中");
                            log.error(e.getMessage());
                            completeQueue.offer(obj);
                        }
                       log.info("CompleteQueueListener任务结束,元素:{}",obj.toString());
                    }
                }
            }
        }).start();
    }
}
利用Springboot的扩展点ApplicationRunner，在项目启动完成后，分别启动BeforeQueueListener、RunningQueueListener、CompleteQueueListener，让三个监听器进入阻塞监听状态
@Component
public class MyRunner implements ApplicationRunner {
    @Autowired
    private ApplicationContext applicationContext;
    @Override
    public void run(ApplicationArguments args) throws Exception {
        Map beansOfType = applicationContext.getBeansOfType(Listener.class);
        for (String s : beansOfType.keySet()) {
            Listener listener = beansOfType.get(s);
            listener.start();
        }


    }
}
结果验证
图片
一个比较有意思的问题
日志丢失的问题
三个任务队列分别有三个线程来进行阻塞监听，即如果任务队列中有任务元素，则取出进行处理；如果没有，则阻塞等待，主线程只负责把任务设置到任务队列中，出现的问题是：控制台的日志输出显示任务元素已经放置到第一个BeforeQueue中，按照预期的结果应该是，控制台的日志输出会显示，从BeforeQueue取出元素进行业务处理、以及业务处理的日志，然后放置到RunningQueue中，再从RunningQueue中取出进行业务处理，接着放置到CompleteQueue队列中，最后从CompleteQueue中取出进行业务处理，最后结束；实际情况是：总是缺少从BeforeQueue取出元素进行业务处理、以及业务处理的日志，其他的日志输出都很正常、执行结果也正常；
问题原因
经过排查分析，最后找到了原因：
是logback线程安全问题， Logback 的大部分组件都是线程安全的，但某些特定的配置可能会导致线程安全问题。例如，如果你在同一个 Appender 中处理多个线程的日志事件，那么可能会出现线程安全问题，导致某些日志事件丢失。
解决方法
问题原因找到了，其实解决方法也就找到，具体就是logback的异步日志，logback.xml配置如下：


    
    
    
    
    
    
        
            ${console.log.pattern}
            utf-8
        
    
    
        500
        0
        true
        
    
    
    
        
    

文章中展示了关键性代码，示例全部代码地址：https://gitcode.net/fox9916/redisson-demo.git
来源：凡夫贬夫内容投诉
免责声明：
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。 
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
软考中级精品资料免费领 
  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 资料下载
  • 历年真题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析（完整版）
     难度      813人已做 
     查看 
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析
     难度      354人已做 
     查看 
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析
     难度      318人已做 
     查看 
  • 2024年上半年软考高项第一、二批次真题考点汇总（完整版）
     难度      435人已做 
     查看 
  • 2024年上半年系统架构设计师考试综合知识真题
     难度      224人已做 
     查看 
相关文章
发现更多好内容
猜你喜欢
AI推送时光机
基于Redis实现消息队列的实践
后端开发Redis消息队列2024-11-30
redis stream 实现消息队列的实践
后端开发2022-08-10
redisstream实现消息队列的实践
后端开发2022-11-13
SpringBoot基于Disruptor实现高效的消息队列 
SpringBoot基于Disruptor实现高效的消息队列 
后端开发2024-04-02
redis实现消息队列的方法
后端开发2024-04-02
redis如何实现消息队列
redis如何实现消息队列
后端开发2024-04-22
redis怎么实现消息队列
后端开发2023-09-14
Redis中如何实现消息队列和延时消息队列
后端开发2024-04-02
Java实现Redis延时消息队列
后端开发2024-04-02
php redis消息队列怎么实现
php redis消息队列怎么实现
后端开发2024-04-11
Redis List 是否适合做消息队列？Spring Boot 与 Redission 实现 Redis 消息队列！
后端开发消息队列Redis2024-11-28
Redis消息队列怎么实现秒杀
后端开发2023-10-11
Redis如何实现消息队列功能
Redis如何实现消息队列功能
后端开发2023-11-07
如何使用redis实现消息队列
后端开发2024-04-02
如何使用 Redis 实现消息队列
后端开发Redis消息队列2024-11-30
Redis中的消息队列序列化怎么实现
Redis中的消息队列序列化怎么实现
后端开发2024-04-29
ThinkPHP基于think-queue的队列插件实现消息推送
后端开发2022-12-14
golang消息队列实现
后端开发2023-05-15
redis用list做消息队列的实现示例
后端开发2024-04-02
Redis消息队列实现的方法是什么
Redis消息队列实现的方法是什么
后端开发2024-04-22
 位置：首页-资讯-后端开发
 咦！没有更多了？去看看其它编程学习网 内容吧 
首页课程
资料下载问答资讯