浅谈三种使用Redis实现MQ的方式

Posted 默辨

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了浅谈三种使用Redis实现MQ的方式相关的知识,希望对你有一定的参考价值。

文章目录





一、消息队列

消息队列(Message Queue),字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色:

  • 消息队列:存储消息
  • 生产者:发送消息到消息队列
  • 消费者:从消息队列获取消息并处理消息


Redis提供了三种不同的方式来实现消息队列:

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





二、基于List的消息队列

Redis的list数据结构是一个双向链表,很容易模拟出队列效果。

队列是入口和出口不在一边,因此我们可以利用:LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP来实现。

不过要注意的是,当队列中没有消息时RPOP或LPOP操作会返回null,并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLOPO来实现阻塞效果。



优点:

足够简单,消费消息延迟几乎为零,但是需要处理空闲连接的问题。基于Redis的持久化机制,数据安全性有保证。基于List可以保证消息的有序性。

如果线程一直阻塞在那里,Redis客户端的连接就成了闲置连接,闲置过久,服务器一般会主动断开连接,减少闲置资源占用,这个时候blpop和brpop或抛出异常,所以在编写客户端消费者的时候要小心,如果捕获到异常需要重试。

缺点:

做消费者确认ACK麻烦,不能保证消费者消费消息后是否成功处理的问题(宕机或处理异常等),通常需要维护一个Pending列表,保证消息处理确认;不能做广播模式,如pub/sub,消息发布/订阅模型;不能重复消费,一旦消费就会被删除;不支持分组消费。





三、基于PubSub的消息队列

PubSub(发布订阅)是Redis2.0版本引入的消息传递模型。顾名思义,消费者可以订阅一个或多个channel,生产者向对应channel发送消息后,所有订阅者都能收到相关消息。

  • SUBSCRIBE channel [channel] :订阅一个或多个频道
  • PUBLISH channel msg :向一个频道发送消息
  • PSUBSCRIBE pattern[pattern] :订阅与pattern格式匹配的所有频道



优点:

典型的广播模式,一个消息可以发布到多个消费者;多信道订阅,消费者可以同时订阅多个信道,从而接收多类消息;消息即时发送,消息不用等待消费者读取,消费者会自动接收到信道发布的消息。

缺点:

消息一旦发布,不能接收。换句话就是发布时若客户端不在线,则消息丢失,不能寻回;不能保证每个消费者接收的时间是一致的;若消费者客户端出现消息积压,到一定程度,会被强制断开,导致消息意外丢失。通常发生在消息的生产远大于消费速度时;可见,Pub/Sub 模式不适合做消息存储,消息积压类的业务,而是擅长处理广播,即时通讯,即时反馈的业务。





四、基于Stream的消息队列

Stream 是 Redis 5.0 引入的一种新数据类型,是一个功能比较完善的消息队列

Stream 的消费模型借鉴了 Kafka 的消费分组的概念,它弥补了 Redis Pub/Sub 不能持久化消息的缺陷。但是它又不同于 kafka,Kafka 的消息可以分 partition,而 Stream 不行。如果非要分 parition 的话,得在客户端做,提供不同的 Stream 名称,对消息进行 hash 取模来选择往哪个 Stream 里塞。



优点:

消息可回溯,一个消息可以被多个消费者读取,可以阻塞读取,有消息确认机制,保证消息至少被消费一次。基本上已经有了一个消息中间件的雏形,可以考虑在生产过程中使用。

缺点:

有消息漏读的风险,并且消息队列的管理和监控就需要花大力气去实现,而专业消息队列都已经自带或者存在着很好的第三方方案和插件。




消费者组(Consumer Group)

将多个消费者划分到一个组中,监听同一个队列,具备下列特点:

  1. 消息分流:队列中的消息会分流给组内的不同消费者,而不是重复消费,从而加快消息处理的速度
  2. 消息标示:消费者组会维护一个标示,记录最后一个被处理的消息,哪怕消费者宕机重启,还会从标示之后读取消息。确保每一个消息都会被消费
  3. 消息确认:消费者获取消息后,消息处于pending状态,并存入一个pending-list。当处理完成后需要通过XACK来确认消息,标记消息为已处理,才会从pending-list移除。



1、基本命令

1、创建消费者组

XGROUP CREATE  key groupName ID [MKSTREAM]
  • key:队列名称
  • groupName:消费者组名称
  • ID:起始ID标示,$代表队列中最后一个消息,0则代表队列中第一个消息
  • MKSTREAM:队列不存在时自动创建队列
# 删除指定的消费者组
XGROUP DESTORY key groupName

# 给指定的消费者组添加消费者
XGROUP CREATECONSUMER key groupname consumername

# 删除消费者组中的指定消费者
XGROUP DELCONSUMER key groupname consumername



2、从消费者组读取消息

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
  • group:消费组名称
  • consumer:消费者名称,如果消费者不存在,会自动创建一个消费者
  • count:本次查询的最大数量
  • BLOCK milliseconds:当没有消息时最长等待时间
  • NOACK:无需手动ACK,获取到消息后自动确认
  • STREAMS key:指定队列名称
  • ID:获取消息的起始ID:
    • “>”:从下一个未消费的消息开始
    • 其它:根据指定id从pending-list中获取已消费但未确认的消息,例如0,是从pending-list中的第一个消息开始




2、简单使用

以下代码案例来自黑马程序员官方Redis教程。

使用背景:使用redis中的lua脚本完成商品下单的操作

1、在lua脚本中完成消息添加到队列中(在Redisson源码中也有这种写法)。即每次完成商品下单之后,会将对应的下单信息封装为一个消息,发送到Redis的名为stream.orders的Stream队列中


2、开启一个线程不断的后台监听对应的队列中的消息

  • 一个死循环任务:用来处理消息,完成Stream队列中消息的消费。出现异常,使用对应的异常策略,即代码中的handlePendingList方法;
  • 一个消息确认机制:消息被消费完成是一定需要一个XACK的确认命令,为确认则认为该消息出现异常,使用异常处理策略;
  • 一个异常处理兜底:用来处理出现错误的消息,该方法中同样需要XACK命令确认后才能说明该消息成功,如果未确认成功,则还是会进入该方法再次消费该消息。
private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();


@PostConstruct
private void init() 
    // 在项目启动后,开启一个后台进程,用来监听stream队列中的消息
    SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());


private class VoucherOrderHandler implements Runnable 

    @Override
    public void run() 
        while (true) 
            try 
                // 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                );
                // 2.判断订单信息是否为空
                if (list == null || list.isEmpty()) 
                    // 如果为null,说明没有消息,继续下一次循环
                    continue;
                
                // 解析数据
                MapRecord<String, Object, Object> record = list.get(0);
                Map<Object, Object> value = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                // 3.创建订单
                createVoucherOrder(voucherOrder);
                // 4.确认消息 XACK
                stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());
             catch (Exception e) 
                log.error("处理订单异常", e);
                handlePendingList();
            
        
    

    // 兜底方法
    private void handlePendingList() 
        while (true) 
            try 
                // 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create("stream.orders", ReadOffset.from("0"))
                );
                // 2.判断订单信息是否为空
                if (list == null || list.isEmpty()) 
                    // 如果为null,说明没有异常消息,结束循环
                    break;
                
                // 解析数据
                MapRecord<String, Object, Object> record = list.get(0);
                Map<Object, Object> value = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                // 3.创建订单
                createVoucherOrder(voucherOrder);
                // 4.确认消息 XACK
                stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());
             catch (Exception e) 
                log.error("处理订单异常", e);
            
        
    





五、总结

总的来说,如果是中小项目和企业,在工作中已经使用了Redis,在业务量不是很大,而又需要消息中间件功能的情况下,可以考虑使用Redis的Stream功能。但是如果并发量很高,资源足够支持下,还是以专业的消息中间件,比如RocketMQ、Kafka等来支持业务更好。

ListPubSubStream
消息持久化支持不支持支持
阻塞读取支持支持支持
消息堆积处理受限于内存空间受限于消费者缓冲区受限于队列长度,可以利用消费者组提高消费速度,减少堆积
消息确认机制不支持不支持支持
消息回溯不支持不支持支持

以上是关于浅谈三种使用Redis实现MQ的方式的主要内容,如果未能解决你的问题,请参考以下文章

浅谈三层

浅谈三层架构MVC之间的关系

salesforce lightning零基础学习 Aura Js 浅谈三: $AActionUtil篇

浅谈js函数三种定义方式 & 四种调用方式 & 调用顺序

浅谈redis——哨兵模式/集群

C#怎么使用redis实现秒杀功能