Redis教程

Redis系列14:使用List实现消息队列

本文主要是介绍Redis系列14:使用List实现消息队列,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

Redis系列1:深刻理解高性能Redis的本质
Redis系列2:数据持久化提高可用性
Redis系列3:高可用之主从架构
Redis系列4:高可用之Sentinel(哨兵模式)
Redis系列5:深入分析Cluster 集群模式
追求性能极致:Redis6.0的多线程模型
追求性能极致:客户端缓存带来的革命
Redis系列8:Bitmap实现亿万级数据计算
Redis系列9:Geo 类型赋能亿级地图位置计算
Redis系列10:HyperLogLog实现海量数据基数统计
Redis系列11:内存淘汰策略
Redis系列12:Redis 的事务机制
Redis系列13:分布式锁实现

1 介绍

在分布式系统中,很重要的一个能力就是消息中间件。我们通过消息队列实现 功能解耦、消息有序性、消息路由、异步处理、流量削峰 等能力。
目前主流的Mq主要有 RabbitMQ 、RocketMQ、kafka,可以参考这篇《MQ系列2:消息中间件技术选型》。
那除了这些主流MQ之外,咱们的这一节要说的Redis也具备实现消息队列的能力。
我们来看看消息队列主要要实现哪些能力,原理是什么,以及如何在 Redission 中应用。

2 关于消息队列

2.1 什么是消息队列

消息中间件是指在分布式系统中完成消息的发送和接收的基础软件。
消息中间件也可以称消息队列(Message Queue / MQ),用高效可靠的消息传递机制进行与平台无关的数据交流,并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息队列模型,可以在分布式环境下扩展进程的通信。
简而言之,互联网场景中经常使用消息中间件进行消息路由、订阅发布、异步处理等操作,来缓解系统的压力。

image

  • Broker: 消息服务器,作为Server提供消息核心服务,一般会包含多个Q。
  • Producer: 消息生产者,业务的发起方,负责生产消息传输给broker,
  • Consumer: 消息消费者,业务的处理方,负责从broker获取消息并进行业务逻辑处理

2.2 它解决了我们哪些问题

1、解耦: 比如说系统A会交给系统B去处理一些事情,但是A不想直接跟B有关联,避免耦合太强,就可以通过在A,B中间加入消息队列,A将要任务的事情交给消息队列 ,B订阅消息队列来执行任务。

这种场景很常见,比如A是订单系统,B是库存系统,可以通过消息队列把削减库存的工作交予B系统去处理。如果A系统同时想让B、C、D...多个系统处理问题的时候,这种优势就更加明显了。

image

2、有序性: 先进先出原理,先来先处理,比如一个系统处理某件事需要很长一段时间,但是在处理这件事情时候,有其他人也发出了请求,可以把请求放在消息队里,一个一个来处理。

对数据的顺序性和一致性有强需求的业务,比如同一张银行卡同时被多个入口使用,需要保证入账出账的顺序性,避免出现数据不一致。

image

3、消息路由: 按照不同的规则,将队列中消息发送到不同的其他队列中

通过消息队列将不同染色的请求发送到不同的服务去操作。这样达成了流量按照业务拆分的目的。

4、异步处理: 处理一项任务的时候,有3个步骤A、B、C,需要先完成A操作, 然后做B、C 操作。任务执行成功与否强依赖A的结果,但不依赖B、C 的结果。
如果我们使用串行的执行方式,那处理任务的周期就会变长,系统的整体吞吐能力也会降低(在同一个系统中做异步其实也是比较大的开销),所以使用消息队列是比较好的办法。

登录操作就是典型的场景:A:执行登录并得到结果、B:记录登录日志、C:将用户信息和Token写入缓存。 执行完A就可以从登录页跳到首页了,B、C让服务慢慢去消化,不阻塞当前操作。

image

5、削峰: 将峰值期间的操作削减,比如A同学的整个操作流程包含12个步骤,后续的11个步骤是不需要强关注结果的数据,可以放在消息队列中。

详细可参考笔者这篇《MQ系列1:消息中间件执行原理》。

2.3 消息队列满足的业务特性

2.3.1 消息有序性

正如上面提到的有序性一样,他能够保证消息按照生产的顺序进行处理和消费,避免消息被无序处理的情况发生。

2.3.2 消息去重

同样的,生产和消费的消息需要保证幂等性原理。避免出现重复执行的情况,
而消息队列的去重机制,也需要确保避免消息被重复消费的问题。

2.3.3 消息的可靠性传输

消息队列的数据可以实现重试、持久化存储、死信队列记录等,以避免消息无法成功传递所产生的不一致现象。
当消息服务器或者消费者恢复健康的时候,可以继续读取消息进行处理,防止消息遗漏。

3 使用Redis的List实现消息队列

稍微学过数据结构都知道。我们经常说Queue(队列),他的存储和使用规则是【先进先出】,栈的存储和使用规则是【先进后出】。
所以List本质上是一个线性的有序结构,也就是Queue的存储关系,它能够保证消费的有序性,按照顺序进行处理。

3.1 入列操作 LPUSH

即进行消息生产,入列操作语法:

 LPUSH key element[element...] 

如果key存在,Producer 通过 LPUSH 将消息插入该队列的头部;如果 key 不存在,则是先创建一个空队列,然后在进行数据插入。
下面举个例子,往队列中插入几个消息,然后得到的返回值是插入消息的个数。

> LPUSH msg_queue msg1 msg2 msg3
(integer) 3

这边往 key 为 msg_queue 的队列中插入了三个消息 msg1、msg2、msg3。

3.2 出列操作 RPOP

即进行消息消费,消费的顺序是先进先出(先生产先消费),出列使用的语法如下:

> RPOP msg_queue
"msg1"
> RPOP msg_queue
"msg2"
> RPOP msg_queue
"msg3"
> RPOP msg_queue
(nil)

都消费完成之后,就是nil了。
image

3.3 消费及时性问题

不同于常规的MQ,具备订阅模式,消费者可以感知到有新的消息生产出来了,再进行消费。
List的问题在于,生产者向队列插入数据的时候,List 并不会主动通知消费者,所以消费者做不到及时消费。
为了保证消费的及时,可能需要做一个心跳包(1秒执行一次),不断地执行 RPOP 指令,当探测到有新消息就会取出消息进行消费,没有消息的时候就返回nil。
但是这种也存在明显的短板,就是不断的调用 RPOP 指令,占用 I/O 资源和CPU资源。

比较好的解决办法就是在队列为空队列的时候,暂停读取,等有消息入列的时候,恢复取数和消费的工作,这样也避免了无效的资源浪费。
Redis 提供了 BLPOP、BRPOP ,无数据的时候自动阻塞读取的命令,有新消息进入的时候,恢复消息取数,如下:

# BRPOP  key  timeout 
BRPOP  msg_queue  0

命令最后一个参数 timeout 是超时时间,单位是秒,如果 timeout 大于0,则到达指定的秒数即使没有弹出成功也会返回,如果 timeout 的值为0,则会一直阻塞等待其他连接向列表中插入元素, timeout 参数不允许为负数。

3.4 消息的重复消费问题

目前 List 没有纯幂等的鉴别能力,但是可以通过以下两种方法来实现:

  • List为每一条消息生成一个 Glocal ID,重复的Glocal ID 不进行重复消费。
  • Producer在生产消息的时候在消息中创建一个Glocal ID,当消费的时候把Glocal ID Record一下,后续的消费先判断再消费,避免重复消费同一个消息。
    这样就保证了对于同一条消息,消费者始终只处理一次,结果始终保持一致。

3.5 消息的可靠性传输问题

可靠性传输我们在MQ篇章用了一整节来介绍持久化存储、消息ACK 、二次记录保障。这边我们也来看看Redis List中的可靠性传输的保障。
Redis中缺少了一个消息确认(ACK)的机制,如果消费数据的时候运行崩溃了,没有确认机制,很可能这条消息就被错过了,无法保证数据的一致性。
解决方案:Redis 提供了 RPOPLPUSH 指令,当List读取消息的时候,会同步的把该消息复制到另外一个List以作备份。
整个操作过程是具备原子性的,避免读取消息了,但是同步备份不成功。

如果出现处理消息出现故障的情况,在故障回复之后,可以从备份的List中复制消息继续消费。操作如下:

# 生产消息 msg1 msg2
> LPUSH list_queue msg1 msg2  
(integer) 2
# 消费消息并同步到备份
> RPOPLPUSH list_queue list_queue_bak
"msg1"
# 当发生故障的时候去消费备份的数据,可以消费到
> RPOP list_queue_bak
"msg1"

如果消费成功则把 list_queue_bak 消息删除即可,如果发生故障,则可以继续从 list_queue_bak 再次读取消息处理。
image

4 使用 Redission 实现队列能力

这边以Java SpringBoot为例子进行说明,可以参考官方文档。

4.1 添加maven依赖 和 配置基本连接

# maven信息
<dependency>
  <groupId>org.redisson</groupId>
  <artifactId>redisson-spring-boot-starter</artifactId>
  <version>3.16.8</version>
</dependency>
# 基本配置
spring:
  application:
    name: redission_test
  redis:
    host: x.x.x.x
    port: 6379
    ssl: false
    password: xxxx.xxxx

4.2 Java程序实现

@Slf4j
@Service
public class RedisQueueService {

    @Autowired
    private RedissonClient redissonClient;

    private static final String REDIS_QUEUE = "listQueue";

    /**
     * 消息生产
     *
     * @param msg
     */
    public void msgProduce(String msg) {
        RBlockingDeque<String> blockDeque = redissonClient.getBlockingDeque(REDIS_QUEUE);
        try {
            blockDeque.putFirst(msg); // 消息写入队列头部
        } catch (InterruptedException e) {
            log.error(e.printStackTrace());
        }
    }

    /**
     * 消息消费:阻塞
     */
    public void msgConsume() {
        RBlockingDeque<String> blockDeque = redissonClient.getBlockingDeque(REDIS_QUEUE);
		Boolen isCheck = true;
        while (isCheck) {
            try {
                String msg = blockDeque.takeLast();  // 从队列中取出消息
            } catch (InterruptedException e) {
                log.error(e.printStackTrace());
            }
        }
    }

5 总结

  • Redis中使用List 数据结构实现消息队列,满足FIFO的处理机制,使用 RPOP 进行消息读取。
  • 使用 BRPOP 指令处理消费及时性问题
  • 使用 BRPOPLPUSH 命令进行消息数据备份,解决消息可靠性传输问题。
  • 相对于专业的MQ,如kafka和RocketMQ,处理能力会差很多。所以在在消息量不大的场景中使用,可以作为一个比较不错的消息队列解决方案。但是过于复杂的场景容易造成消息堆积。
这篇关于Redis系列14:使用List实现消息队列的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!