[大数据] Redis消息队列发展历程

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> Redis消息队列发展历程 -> 正文阅读

[大数据]Redis消息队列发展历程

Redis是目前最受欢迎的kv类数据库，当然它的功能越来越多，早已不限定在kv场景，消息队列就是Redis中一个重要的功能。

Redis从2010年发布1.0版本就具备一个消息队列的雏形，随着10多年的迭代，其消息队列的功能也越来越完善，作为一个全内存的消息队列，适合应用与要求高吞吐、低延时的场景。

我们来盘一下Redis消息队列功能的发展历程，历史版本有哪些不足，后续版本是如何来解决这些问题的。

一 Redis 1.0 list

从广义上来讲消息队列就是一个队列的数据结构，生产者从队列一端放入消息，消费者从另一端读取消息，消息保证先入先出的顺序，一个本地的list数据结构就是一个进程维度的消息队列，它可以让模块A写入消息，模块B消费消息，做到模块A/B的解耦与异步化。但想要做到应用级别的解耦和异步还需要一个消息队列的服务。

1 list的特性

Redis 1.0发布时就具备了list数据结构，应用A可以通过lpush写入消息，应用B通过rpop从队列中读取消息，每个消息只会被读取一次，而且是按照lpush写入的顺序读到。同时Redis的接口是并发安全的，可以同时有多个生产者向一个list中生产消息，多个消费者从list中读取消息。

这里还有个问题，消费者要如何知道list中有消息了，需要不断轮询去查询吗。轮询无法保证消息被及时的处理，会增加延时，而且当list为空时，大部分轮询的请求都是无效请求，这种方式大量浪费了系统资源。好在Redis有brpop接口，该接口有一个参数是超时时间，如果list为空，那么Redis服务端不会立刻返回结果，它会等待list中有新数据后在返回或是等待最多一个超时时间后返回空。通过brpop接口实现了长轮询，该效果等同于服务端推送，消费者能立刻感知到新的消息，而且通过设置合理的超时时间，使系统资源的消耗降到很低。

#基于list完成消息的生产和消费

#生产者生产消息msg1
lpush listA msg1
(integer) 1

#消费者读取到消息msg1
rpop listA
"msg1"

#消费者阻塞式读取listA，如果有数据立刻返回，否则最多等待10秒
brpop listA 10 
1) "listA"
2) "msg1"

使用rpop或brpop这样接口消费消息会先从队列中删除消息，然后再由应用消费，如果应用应用在处理消息前异常宕机了，消息就丢失了。但如果使用lindex这样的只读命令先读取消息处理完毕后在删除，又需要额外的机制来保证一条消息不会被其他消费者重复读到。好在list有rpoplpush或brpoplpush这样的接口，可以原子性的从一个list中移除一个消息并加入另一个list。

应用程序可以通过2个list组和来完成消息的消费和确认功能，使用rpoplpush从list A中消费消息并移入list B，等消息处理完毕后在从list B中删除消息，如果在处理消息过程中应用异常宕机，恢复后应用可以重新从list B中读取未处理的消息并处理。这种方式为消息的消费增加了ack机制。

#基于2个list完成消息消费和确认

#从listA中读取消息并写入listB
rpoplpush listA listB
"msg1"

#业务逻辑处理msg1完毕后，从listB中删除msg1，完成消息的确认
lrem listB 1 msg1
(integer) 1

2 list的不足之处

通过Redis 1.0就引入的list结构我们就能实现一个分布式的消息队列，满足一些简单的业务需求。但list结构作为消息队列服务有一个很致命的问题，它没有广播功能，一个消息只能被消费一次。而在大型系统中，通常一个消息会被下游多个应用同时订阅和消费，例如当用户完成一个订单的支付操作时，需要通知商家发货，要更新物流状态，可能还会提高用户的积分和等级，这些都是不同的下游子系统，他们全部会订阅支付完成的操作，而list一个消息只能被消费一次在这样复杂的大型系统面前就捉襟见肘了。

可能你会说那弄多个list，生产者向每个list中都投递消息，每个消费者处理自己的list不就行了吗。这样第一是性能不会太好，因为同一个消息需要被重复的投递，第二是这样的设计违反了生产者和消费者解耦的原则，这个设计下生产者需要知道下游有哪些消费者，如果业务发生变化，需要额外增加一个消费者，生产者的代码也需要修改。

3 总结

优势

模型简单，和使用本地list基本相同，适配容易
通过brpop做到消息处理的实时性
通过rpoplpush来联动2个list，可以做到消息先消费后确认，避免消费者应用异常情况下消息丢失

不足

消息只能被消费一次，缺乏广播机制

二 Redis 2.0 pubsub

list作为消息队列应用场景受到限制很重要的原因在于没有广播，所以Redis 2.0中引入了一个新的数据结构pubsub。pubsub虽然不能算作是list的替代品，但它确实能解决一些list不能解决的问题。

1 pubsub特性

pubsub引入一个概念叫channel，生产者通过publish接口投递消息时会指定channel，消费者通过subscribe接口订阅它关心的channel，调用subscribe后这条连接会进入一个特殊的状态，通常不能在发送其他请求，当有消息投递到这个channel时Redis服务端会立刻通过该连接将消息推送到消费者。这里一个channel可以被多个应用订阅，消息会同时投递到每个订阅者，做到了消息的广播。

另一方面，消费者可以会订阅一批channel，例如一个用户订阅了浙江的新闻的推送，但浙江新闻还会进行细分，例如“浙江杭州xx”、“浙江温州xx”，这里订阅者不需要获取浙江的所有子类在挨个订阅，只需要调用psubscribe“浙江*”就能订阅所有以浙江开头的新闻推送了，这里psubscribe传入一个通配符表达的channel，Redis服务端按照规则推送所有匹配channel的消息给对应的客户端。

#基于pubsub完成channel的匹配和消息的广播

#消费者1订阅channel1
subscribe channel1
1) "subscribe"
2) "channel1"
3) (integer) 1
#收到消息推送
1) "message"
2) "channel1"
3) "msg1"

#消费者2订阅channel*
psubscribe channel*
1) "psubscribe"
2) "channel*"
3) (integer) 1
#收到消息推送
1) "pmessage"
2) "channel*"
3) "channel1"
4) "msg1"
1) "pmessage"
2) "channel*"
3) "channel2"
4) "msg2"

#生产者发布消息msg1和msg2
publish channel1 msg1
(integer) 2
publish channel2 msg2
(integer) 1

在Redfis 2.8时加入了keyspace notifications功能，此时pubsub除了通知用户自定义消息，也可以通知系统内部消息。keyspace notifications引入了2个特殊的channel分别是__keyevent@__:和__keyspace@__:，通过订阅__keyevent客户端可以收到某个具体命令调用的回调通知，通过订阅__keyspace客户端可以收到目标key的增删改操作以及过期事件。使用这个功能还需要开启配置notify-keyspace-events。

#通过keyspace notifications功能获取系统事件

#写入请求
set testkey v EX 1

#订阅key级别的事件
psubscribe __keyspace@0__:testkey
1) "psubscribe"
2) "__keyspace@0__:testkey"
3) (integer) 1
#收到通知
1) "pmessage"
2) "__keyspace@0__:testkey"
3) "__keyspace@0__:testkey"
4) "set"
1) "pmessage"
2) "__keyspace@0__:testkey"
3) "__keyspace@0__:testkey"
4) "expire"
1) "pmessage"
2) "__keyspace@0__:testkey"
3) "__keyspace@0__:testkey"
4) "expired"

#订阅所有的命令事件
psubscribe __keyevent@0__:*
1) "psubscribe"
2) "__keyevent@0__:*"
3) (integer) 1
#收到通知
1) "pmessage"
2) "__keyevent@0__:*"
3) "__keyevent@0__:set"
4) "testkey"
1) "pmessage"
2) "__keyevent@0__:*"
3) "__keyevent@0__:expire"
4) "testkey"
1) "pmessage"
2) "__keyevent@0__:*"
3) "__keyevent@0__:expired"
4) "testkey"

2 pubsub的不足之处

pubsub既能单播又能广播，还支持channel的简单正则匹配，功能上已经能满足大部分业务的需求，而且这个接口发布的时间很早，在2011年Redis 2.0发布时就已经具备，用户基础很广泛，所以现在很多业务都有用到这个功能。但你要深入了解pubsub的原理后，是肯定不敢把它作为一个一致性要求较高，数据量较大系统的消息服务的。

首先，pubsub的消息数据是瞬时的，它在Redis服务端不做保存，publish发送到Redis的消息会立刻推送到所有当时subscribe连接的客户端，如果当时客户端因为网络问题断连，那么就会错过这条消息，当客户端重连后，它没法重新获取之前那条消息，甚至无法判断是否有消息丢失。

其次，pubsub中消费者获取消息是一个推送模型，这意味着Redis会按消息生产的速度给所有的消费者推送消息，不管消费者处理能力如何，如果消费者应用处理能力不足，消息就会在Redis的client buf中堆积，当堆积数据超过一个阈值后会断开这条连接，这意味着这些消息全部丢失了，在也找不回来了。如果同时有多个消费者的client buf堆积数据但又还没达到断开连接的阈值，那么Redis服务端的内存会膨胀，进程可能因为oom而被杀掉，这导致了整个服务中断。

3 总结

优势

消息具备广播能力
psubscribe能按字符串通配符匹配，给予了业务逻辑的灵活性
能订阅特定key或特定命令的系统消息

不足

Redis异常、客户端断连都会导致消息丢失
消息缺乏堆积能力，不能削峰填谷。推送的方式缺乏背压机制，没有考虑消费者处理能力，推送的消息超过消费者处理能力后可能导致消息丢失或服务异常

三 Redis 5.0 stream

消息丢失、消息服务不稳定的问题严重限制了pubsub的应用场景，所以Redis需要重新设计一套机制，来解决这些问题，这就有了后来的stream结构。

1 stream特性

一个稳定的消息服务需要具备几个要点，要保证消息不会丢失，至少被消费一次，要具备削峰填谷的能力，来匹配生产者和消费者吞吐的差异。在2018年Redis 5.0加入了stream结构，这次考虑了list、pubsub在应用场景下的缺陷，对标kafka的模型重新设计全内存消息队列结构，从这时开始Redis消息队列功能算是能和主流消息队列产品pk一把了。

stream的改进分为多个方面

成本：

存储message数据使用了listpack结构，这是一个紧凑型的数据结构，不同于list的双向链表每个节点都要额外占用2个指针的存储空间，这使得小msg情况下stream的空间利用率更高。

功能：

stream引入了消费者组的概念，一个消费者组内可以有多个消费者，同一个组内的消费者共享一个消息位点（last_delivered_id），这使得消费者能够水平的扩容，可以在一个组内加入多个消费者来线性的提升吞吐，对于一个消费者组，每条msg只会被其中一个消费者获取和处理，这是pubsub的广播模型不具备的。
不同消费者组之前是相互隔离的，他们各自维护自己的位点，这使得一条msg能被多个不同的消费者组重复消费，做到了消息广播的能力。
stream中消费者采用拉取的方式，并能设置timeout在没有消息时阻塞，通过这种长轮询机制保证了消息的实时性，而且消费速率是和消费者自身吞吐相匹配。

消息不丢失：

stream的数据会存储在aof和rdb文件中，这使Redis重启后能够恢复stream的数据。而pubsub的数据是瞬时的，Redis重启意味着消息全部丢失。
stream中每个消费者组会存储一个last_delivered_id来标识已经读取到的位点，客户端连接断开后重连还是能从该位点继续读取，消息不会丢失。
stream引入了ack机制保证消息至少被处理一次。考虑一种场景，如果消费者应用已经读取了消息，但还没来得及处理应用就宕机了，对于这种已经读取但没有ack的消息，stream会标示这条消息的状态为pending，等客户端重连后通过xpending命令可以重新读取到pengind状态的消息，继续处理。如果这个应用永久宕机了，那么该消费者组内的其他消费者应用也能读取到这条消息，并通过xclaim命令将它归属到自己下面继续处理。

#基于stream完成消息的生产和消费，并确保异常状态下消息至少被消费一次

#创建mystream，并且创建一个consumergroup为mygroup
XGROUP CREATE mystream mygroup $ MKSTREAM
OK

#写入一条消息，由redis自动生成消息id，消息的内容是一个kv数组，这里包含field1 value1 field2 value2
XADD mystream * field1 value1 field2 value2
"1645517760385-0"

#消费者组mygroup中的消费者consumer1从mystream读取一条消息，>表示读取一条该消费者组从未读取过的消息
XREADGROUP GROUP mygroup consumer1 COUNT 1 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) "1645517760385-0"
         2) 1) "field1"
            2) "value1"
            3) "field2"
            4) "value2"
            
#消费完成后ack确认消息
xack mystream mygroup 1645517760385-0
(integer) 1

#如果消费者应用在ack前异常宕机，恢复后重新获取未处理的消息id。
XPENDING mystream mygroup - + 10 
1) 1) "1645517760385-0"
   2) "consumer1"
   3) (integer) 305356
   4) (integer) 1
   
 #如果consumer1永远宕机，其他消费者可以把pending状态的消息移动到自己名下后继续消费
 #将消息id 1645517760385-0移动到consumer2下
 XCLAIM mystream mygroup consumer2 0 1645517760385-0
 1) 1) "1645517760385-0"
   2) 1) "field1"
      2) "value1"
      3) "field2"
      4) "value2"

Redis stream保证了消息至少被处理一次，但如果想做到每条消息仅被处理一次还需要应用逻辑的介入。

消息被重复处理要么是生产者重复投递，要么是消费者重复消费。

对于生产者重复投递问题，Redis stream为每个消息都设置了一个唯一递增的id，通过参数可以让Redis自动生成id或者应用自己指定id，应用可以根据业务逻辑为每个msg生成id，当xadd超时后应用并不能确定消息是否投递成功，可以通过xread查询该id的消息是否存在，存在就说明已经投递成功，不存在则重新投递，而且stream限制了id必须递增，这意味了已经存在的消息重复投递会被拒绝。这套机制保证了每个消息可以仅被投递一次。
对于消费者重复消费的问题，考虑一个场景，消费者读取消息后业务处理完毕，但还没来得及ack就发生了异常，应用恢复后对于这条没有ack的消息进行了重复消费。这个问题因为ack和消费消息的业务逻辑发生在2个系统，没法做到事务性，需要业务来改造，保证消息处理的幂等性。

2 stream的不足

stream的模型做到了消息的高效分发，而且保证了消息至少被处理一次，通过应用逻辑的改造能做到消息仅被处理一次，它的能力对标kafka，但吞吐高于kafka，在高吞吐场景下成本比kafka低，那它又有哪些不足了。

首先消息队列很重要的一个功能就是削峰填谷，来匹配生产者和消费者吞吐的差异，生产者和消费者吞吐差异越大，持续时间越长，就意味着steam中需要堆积更多的消息，而Redis作为一个全内存的产品，数据堆积的成本比磁盘高。

其次stream通过ack机制保证了消息至少被消费一次，但这有个前提就是存储在Redis中的消息本身不会丢失。Redis数据的持久化依赖aof和rdb文件，aof落盘方式有几种，通过配置appendfsync决定，通常我们不会配置为always来让每条命令执行完后都做一次fsync，线上配置一般为everysec，每秒做一次fsync，而rdb是全量备份时生成，这意味了宕机恢复可能会丢掉最近一秒的数据。另一方面线上生产环境的Redis都是高可用架构，当主节点宕机后通常不会走恢复逻辑，而是直接切换到备节点继续提供服务，而Redis的同步方式是异步同步，这意味着主节点上新写入的数据可能还没同步到备节点，在切换后这部分数据就丢失了。所以在故障恢复中Redis中的数据可能会丢失一部分，在这样的背景下无论stream的接口设计的多么完善，都不能保证消息至少被消费一次。

3 总结

优势

在成本、功能上做了很多改进，支持了紧凑的存储小消息、具备广播能力、消费者能水平扩容、具备背压机制
通过ack机制保证了Redis服务端正常情况下消息至少被处理一次的能力

不足

内存型消息队列，数据堆积成本高
Redis本身rpo>0，故障恢复可能会丢数据，所以stream在Redis发生故障恢复后也不能保证消息至少被消费一次。

四 Tair持久内存版 stream

Redis stream的不足也是内存型数据库特性带来的，它拥有高吞吐、低延时，但大容量下成本会比较高，而应用的场景也不完全是绝对的大容量低吞吐或小容量高吞吐，有时应用的场景会介于二者之间，需要平衡容量和吞吐的关系，所以需要一个产品它的存储成本低于Redis stream，但它的性能又高于磁盘型消息队列。

另一方面Redis stream在Redis故障场景下不能保证消息的不丢失，这导致业务需要自己实现一些复杂的机制来回补这段数据，同时也限制了它应用在一些对一致性要求较高的场景。为了让业务逻辑更简单，stream应用范围更广，需要保证故障场景下的消息持久化。

兼顾成本、性能、持久化，这就有了Tair持久内存版。

1 Tair持久内存版特性

更大空间，更低成本

Tair持久内存版引入了Intel傲腾持久内存（下面称作AEP），它的性能略低于内存，但相同容量下成本低于内存。Tair持久内存版将主要数据存储在AEP上，使得相同容量下，成本更低，这使同样单价下stream能堆积更多的消息。

兼容社区版

Tair持久内存版兼容原生Redis绝大部分的数据结构和接口，对于stream相关接口做到了100%兼容，如果你之前使用了社区版stream，那么不需要修改任何代码，只需要换一个连接地址就能切换到持久内存版。并且通过工具完成社区版和持久内存版数据的双向迁移。

数据的实时持久化

Tair持久内存版并不是简单将Redis中的数据换了一个介质存储，因为这样仅能通过AEP降低成本，但没用到AEP断电数据不丢失的特性，对持久化能力没有任何提升。

开源Redis通过在磁盘上记录AppendOnlyLog来持久化数据，AppendOnlyLog记录了所有的写操作，相当于redolog，在宕机恢复时通过回放这些log恢复数据。但受限于磁盘介质的高延时和Redis内存数据库使用场景下对低延时的要求，并不能在每次写操作后fsync持久化log，最新写入的数据可能并没有持久化到磁盘，这也是数据可能丢失的根因。

Tair持久内存版的数据恢复没有使用AppendOnlyLog来完成，而是将将redis数据结构存储在AEP上，这样宕机后这些数据结构并不会丢失，并且对这些数据结构增加了一些额外的描述信息，宕机后在recovery时能够读到这些额外的描述信息，让这些redis数据结构重新被识别和索引，将状态恢复到宕机前的样子。Tair通过将redis数据结构和描述信息实时写入AEP，保证了写入数据的实时持久化。

HA数据不丢失

Tair持久内存版保证了数据的持久化，但生产环境中都是高可用架构，多数情况下当主节点异常宕机后并不会等主节点重启恢复，而是切换到备节点继续提供服务，然后给新的主节点添加一个新的备节点。所以在故障发生时如果有数据还没从主节点同步到备节点，这部分数据就会丢失。

Redis采用的异步同步，当客户端写入数据并返回成功时对Redis的修改可能还没同步到备节点，如果此时主节点宕机数据就会丢失。为了避免在HA过程中数据丢失，Tair持久内存版引入了半同步机制，确保写入请求返回成功前相关的修改已经同步到备节点。

可以发现开启半同步功能后写入请求的RT会变高，多出主备同步的耗时，这部分耗时大概在几十微秒。但通过一些异步化的技术，虽然写请求的RT会变高，但对实例的最大写吞吐影响很小。

当开启半同步后生成者通过xadd投递消息，如果返回成功，消息一定同步到备节点，此时发生HA，消费者也能在备节点上读到这条消息。如果xadd请求超时，此时消息可能同步到备节点也可能没有，生产者没法确定，此时通过再次投递消息，可以保证该消息至少被消费一次。如果要严格保证消息仅被消费一次，那么生产者可以通过xread接口查询消息是否存在，对于不存在的场景重新投递。

2 总结

优势