Redis 使用 List 实现消息队列的优缺点

目录
  • 什么是消息队列
  • 消息队列满足哪些特性
    • 消息有序性
    • 重复消息处理
    • 可靠性
  • List 实现消息队列
    • LPUSH
    • RPOP
  • 实时消费问题
  • 重复消费
  • 消息可靠性
    • 需要注意的是
  • Redission 实战
  • 添加依赖
  • Java 代码实战
  • 单元测试
  • 总结
    • 需要注意的是,我们要避免生产者过快,消费者过慢导致的消息堆积占用 Redis 的内存。

分布式系统中必备的一个中间件就是消息队列,通过消息队列我们能对服务间进行异步解耦、流量消峰、实现最终一致性。

目前市面上已经有 RabbitMQ、RochetMQ、ActiveMQ、Kafka 等,有人会问:“Redis 适合做消息队列么?”

在回答这个问题之前,我们先从本质思考:

  • 消息队列提供了什么特性?
  • Redis 如何实现消息队列?是否满足存取需求?

今天,码哥结合消息队列的特点一步步带大家分析使用 Redis 的 List 作为消息队列的实现原理,并分享如何把 SpringBoot 与 Redission 整合运用到项目中。

什么是消息队列

消息队列是一种异步的服务间通信方式,适用于分布式和微服务架构。消息在被处理和删除之前一直存储在队列上。

每条消息仅可被一位用户处理一次。消息队列可被用于分离重量级处理、缓冲或批处理工作以及缓解高峰期工作负载。

  • Producer:消息生产者,负责产生和发送消息到 Broker;
  • Broker:消息处理中心。负责消息存储、确认、重试等,一般其中会包含多个 queue;
  • Consumer:消息消费者,负责从 Broker 中获取消息,并进行相应处理;
  • 消息队列的使用场景有哪些呢?
  • 消息队列在实际应用中包括如下四个场景:
  • 应用耦合:发送方、接收方系统之间不需要了解双方,只需要认识消息。多应用间通过消息队列对同一消息进行处理,避免调用接口失败导致整个过程失败;
  • 异步处理:多应用对消息队列中同一消息进行处理,应用间并发处理消息,相比串行处理,减少处理时间;
  • 限流削峰:广泛应用于秒杀或抢购活动中,避免流量过大导致应用系统挂掉的情况;
  • 消息驱动的系统:系统分为消息队列、消息生产者、消息消费者,生产者负责产生消息,消费者(可能有多个)负责对消息进行处理;

消息队列满足哪些特性

消息有序性

消息是异步处理的,但是消费者需要按照生产者发送消息的顺序来消费,避免出现后发送的消息被先处理的情况。

重复消息处理

生产者可能因为网络问题出现消息重传导致消费者可能会收到多条重复消息。

同样的消息重复多次的话可能会造成一业务逻辑多次执行,需要确保如何避免重复消费问题。

可靠性

一次保证消息的传递。如果发送消息时接收者不可用,消息队列会保留消息,直到成功地传递它。

当消费者重启后,可以继续读取消息进行处理,防止消息遗漏。

List 实现消息队列

Redis 的列表(List)是一种线性的有序结构,可以按照元素被推入列表中的顺序来存储元素,能满足「先进先出」的需求,这些元素既可以是文字数据,又可以是二进制数据。

LPUSH

生产者使用 LPUSH key element[element...] 将消息插入到队列的头部,如果 key 不存在则会创建一个空的队列再插入消息。

如下,生产者向队列 queue 先后插入了 「Java」「码哥字节」「Go」,返回值表示消息插入队列后的个数。

> LPUSH queue Java 码哥字节 Go
(integer) 3

RPOP

消费者使用 RPOP key 依次读取队列的消息,先进先出,所以 「Java」会先读取消费:

> RPOP queue
"Java"
> RPOP queue
"码哥字节"
> RPOP queue
"Go"

实时消费问题

65 哥:这么简单就实现了么?

别高兴的太早, LPUSH、RPOP 存在一个性能风险,生产者向队列插入数据的时候,List 并不会主动通知消费者及时消费。

我们需要写一个 while(true) 不停地调用 RPOP 指令,当有新消息就会返回消息,否则返回空。

程序需要不断轮询并判断是否为空再执行消费逻辑,这就会导致即使没有新消息写入到队列,消费者也要不停地调用 RPOP 命令占用 CPU 资源。

65 哥:要如何避免循环调用导致的 CPU 性能损耗呢?

Redis 提供了 BLPOP、BRPOP 阻塞读取的命令, 消费者在在读取队列没有数据的时候自动阻塞,直到有新的消息写入队列,才会继续读取新消息执行业务逻辑。

BRPOP queue 0

参数 0 表示阻塞等待时间无无限制

重复消费

  • 消息队列为每一条消息生成一个「全局 ID」;
  • 生产者为每一条消息创建一条「全局 ID」,消费者把一件处理过的消息 ID 记录下来判断是否重复。

其实这就是幂等,对于同一条消息,消费者收到后处理一次的结果和多次的结果是一致的。

消息可靠性

65 哥:消费者从 List 中读取一条在消息处理过程中宕机了就会导致消息没有处理完成,可是数据已经没有保存在 List 中了咋办?

本质就是消费者在处理消息的时候崩溃了,就无法再还原消息,缺乏一个消息确认机制。

Redis 提供了 RPOPLPUSH、BRPOPLPUSH(阻塞) 两个指令,含义是从 List 从读取消息的同时把这条消息复制到另一个 List 中(备份),并且是原子操作。

我们就可以在业务流程正确处理完成后再删除队列消息实现消息确认机制。如果在处理消息的时候宕机了,重启后再从备份 List 中读取消息处理。

LPUSH redisMQ 公众号 码哥字节
BRPOPLPUSH redisMQ redisMQBack

生产者用 LPUSH 把消息插入到 redisMQ 队列中,消费者使用 BRPOPLPUSH 读取消息「公众号」,同时该消息会被插入到 「redisMQBack」队列中。

如果消费成功则把「redisMQBack」的消息删除即可,异常的话可以继续从 「redisMQBack」再次读取消息处理。

需要注意的是

如果生产者消息发送的很快,而消费者处理速度慢就会导致消息堆积,给 Redis 的内存带来过大压力。

Redission 实战

在 Java 中,我们可以利用 Redission 封装的 API 来快速实现队列,接下来码哥基于 SpringBoot 2.1.4 版本来交大家如何整合并实战。

详细 API 文档大家可查阅: https://github.com/redisson/redisson/wiki/7.-Distributed-collections

添加依赖

<dependency>
  <groupId>org.redisson</groupId>
  <artifactId>redisson-spring-boot-starter</artifactId>
  <version>3.16.7</version>
</dependency>

添加 Redis 配置,码哥的 Redis 没有配置密码,大家根据实际情况配置即可。

spring:
  application:
    name: redission
  redis:
    host: 127.0.0.1
    port: 6379
    ssl: false

Java 代码实战

RBlockingDeque 继承 java.util.concurrent.BlockingDeque ,在使用过程中我们完全可以根据接口文档来选择合适的 API 去实现业务逻辑。

主要方法如下

码哥采用了双端队列来举例

@Slf4j
@Service
public class QueueService {

    @Autowired
    private RedissonClient redissonClient;

    private static final String REDIS_MQ = "redisMQ";

    /**
     * 发送消息到队列头部
     *
     * @param message
     */
    public void sendMessage(String message) {
        RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque(REDIS_MQ);

        try {
            blockingDeque.putFirst(message);
            log.info("将消息: {} 插入到队列。", message);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 从队列尾部阻塞读取消息,若没有消息,线程就会阻塞等待新消息插入,防止 CPU 空转
     */
    public void onMessage() {
        RBlockingDeque<String> blockingDeque = redissonClient.getBlockingDeque(REDIS_MQ);
        while (true) {
            try {
                String message = blockingDeque.takeLast();
                log.info("从队列 {} 中读取到消息:{}.", REDIS_MQ, message);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }
    }

单元测试

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RedissionApplication.class)
public class RedissionApplicationTests {
    @Autowired
    private QueueService queueService;
    @Test
    public void testQueue() throws InterruptedException {
        new Thread(() -> {
            for (int i = 0; i < 1000; i++) {
                queueService.sendMessage("消息" + i);
            }
        }).start();
        new Thread(() -> queueService.onMessage()).start();
        Thread.currentThread().join();
    }
}

总结

可以使用 List 数据结构来实现消息队列,满足先进先出。为了实现消息可靠性,Redis 提供了 BRPOPLPUSH 命令是解决。

Redis 是一个非常轻量级的键值数据库,部署一个 Redis 实例就是启动一个进程,部署 Redis 集群,也就是部署多个 Redis 实例。

而 Kafka、RabbitMQ 部署时,涉及额外的组件,例如 Kafka 的运行就需要再部署 ZooKeeper。相比 Redis 来说,Kafka 和 RabbitMQ 一般被认为是重量级的消息队列。

需要注意的是,我们要避免生产者过快,消费者过慢导致的消息堆积占用 Redis 的内存。

在消息量不大的情况下使用 Redis 作为消息队列,他能给我们带来高性能的消息读写,这似乎也是一个很好消息队列解决方案。

到此这篇关于Redis 使用 List 实现消息队列有哪些利弊的文章就介绍到这了,更多相关redis list消息队列内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

时间: 2022-01-08

基于Redis的List实现特价商品列表功能

目录 1.场景分析 2.分析 3 .具体实现 3.1 ProductListService类 3.2 商品的数据接口的定义和展示及分页 3.3 定时任务 4.解决商品列表存在的缓存击穿问题 4.1 如何引起的缓存击穿的情况 4.2 解决方案 1.场景分析 淘宝京东的特价商品列表, 商品特点: 商品有限,并发量非常的大. 考虑分页 传统解决方案:数据库db, 但是在如此大的并发量的情况下,不可取. 一般会采用redis来处理.这些特价商品的数据不多,而且redis的list本身也支持分页.是天然处

使用 Redis 流实现消息队列的代码

在介绍了 Redis 流的基本功能之后, 现在是时候使用这些功能来构建一些实际的应用了. 消息队列作为流的典型应用之一, 具有非常好的示范性, 因此我们将使用 Redis 流的相关功能构建一个消息队列应用, 这个消息队列跟我们之前使用其他 Redis 数据结构构建的消息队列具有相似的功能. 代码清单 10-1 展示了一个具有基本功能的消息队列实现: 代码最开头的是几个转换函数, 它们负责对程序的相关输入输出进行转换和格式化: MessageQueue 类用于实现消息队列, 它的添加消息.移除消息

golang实现redis的延时消息队列功能示例

前言 在学习过程中发现redis的zset还可以用来实现轻量级的延时消息队列功能,虽然可靠性还有待提高,但是对于一些对数据可靠性要求不那么高的功能要求完全可以实现.本次主要采用了redis中zset中的zadd, zrangebyscore 和 zdel来实现一个小demo. 提前准备 安装redis, redis-go 因为用的是macOS, 直接 $ brew install redis $ go get github.com/garyburd/redigo/redis 又因为比较懒,生成任

PHP使用php-resque库配合Redis实现MQ消息队列的教程

消息队列处理后台任务带来的问题 项目中经常会有后台运行任务的需求,比如发送邮件时,因为要连接邮件服务器,往往需要5-10秒甚至更长时间,如果能先给用户一个成功的提示信息,然后在后台慢慢处理发送邮件的操作,显然会有更好的用户体验. 为了实现类似的需求,Web项目中一般的实现方法是使用消息队列(Message Queue),比如MemcacheQ,RabbitMQ等等,都是很著名的产品. 消息队列说白了就是一个最简单的先进先出队列,队列的一个成员就是一段文本.正是因为消息队列实在太简单了,当拿着消息

深入理解redis分布式锁和消息队列

最近博主在看redis的时候发现了两种redis使用方式,与之前redis作为缓存不同,利用的是redis可设置key的有效时间和redis的BRPOP命令. 分布式锁 由于目前一些编程语言,如PHP等,不能在内存中使用锁,或者如Java这样的,需要一下更为简单的锁校验的时候,redis分布式锁的使用就足够满足了. redis的分布式锁其实就是基于setnx方法和redis对key可设置有效时间的功能来实现的.基本用法比较简单. public boolean tryLock(String loc

PHP+memcache实现消息队列案例分享

memche消息队列的原理就是在key上做文章,用以做一个连续的数字加上前缀记录序列化以后消息或者日志.然后通过定时程序将内容落地到文件或者数据库. php实现消息队列的用处比如在做发送邮件时发送大量邮件很费时间的问题,那么可以采取队列.方便实现队列的轻量级队列服务器是:starling支持memcache协议的轻量级持久化服务器https://github.com/starling/starlingBeanstalkd轻量.高效,支持持久化,每秒可处理3000左右的队列http://kr.gi

.net msmq消息队列实例详解

本文为大家分享了.net msmq消息队列实例代码,供大家参考,具体内容如下 1.msmq消息队列windows环境安装 控制面板---->程序和功能---->启用或关闭Windows程序---->Microsoft Message Queue(MSMQ)服务器 选中如图所示功能点击"确认"进行安装,安装好后可在 "计算机管理"中进行查看 2.创建消息队列实体对象 /// <summary> /// 消息实体 /// </summ

PHP高级编程之消息队列原理与实现方法详解

本文实例讲述了PHP高级编程之消息队列原理与实现方法.分享给大家供大家参考,具体如下: 1. 什么是消息队列 消息队列(英语:Message queue)是一种进程间通信或同一进程的不同线程间的通信方式 2. 为什么使用消息队列 消息队列技术是分布式应用间交换信息的一种技术.消息队列可驻留在内存或磁盘上,队列存储消息直到它们被应用程序读出.通过消息队列,应用程序可独立地执行,它们不需要知道彼此的位置.或在继续执行前不需要等待接收程序接收此消息. 3. 什么场合使用消息队列 你首先需要弄清楚,消息

Java利用Redis实现消息队列的示例代码

本文介绍了Java利用Redis实现消息队列的示例代码,分享给大家,具体如下: 应用场景 为什么要用redis? 二进制存储.java序列化传输.IO连接数高.连接频繁 一.序列化 这里编写了一个java序列化的工具,主要是将对象转化为byte数组,和根据byte数组反序列化成java对象; 主要是用到了ByteArrayOutputStream和ByteArrayInputStream; 注意:每个需要序列化的对象都要实现Serializable接口; 其代码如下: package Utils

详解Redis用链表实现消息队列

前言 Redis链表经常会被用于消息队列的服务,以完成多程序之间的消息交换.个人认为redis消息队列有一个好处,就是可以实现分布式和共享,就和memcache作为mysql的缓存和mysql自带的缓存一样. 链表实现消息队列 Redis链表支持前后插入以及前后取出,所以如果往尾部插入元素,往头部取出元素,这就是一种消息队列,也可以说是消费者/生产者模型.可以利用lpush和rpop来实现.但是有一个问题,如果链表中没有数据,那么消费者将要在while循环中调用rpop,这样以来就浪费cpu资源

PHP+Redis 消息队列 实现高并发下注册人数统计的实例

前言 现在越来越多的网站开始注重统计和用户行为分析,作为网站经常使用的功能,如何让统计性能更加高,这也是我们需要考虑的事情.本篇通过Redis来优化统计功能(以注册人数统计为例). 传统的统计功能都是直接操作数据库把数据插入表中.这样做,对数据库的性能消耗就会比较大. 思路: 这里我们用到了redis的队列,注册的时候先添加到队列,然后在处理的时候出队,并且把人数添加redis里. 代码: <?php //register.php $redis = new Redis(); $redis->c