基于Redis实现消息队列的6种方案之方案简述（下）Redis5.0的新类型Stream

不太灵光的程序员 2020-06-29

149

点击蓝色“不太灵光的程序员”关注我哟

加个“星标”，每天上午 08:00，干货推送！

Redis5.0才有的新数据类型Stream基本上可以满足你对消息队列的所有需求，以下的内容有点长，如果你还在忙的话建议先收藏下。

基于Stream实现的消息队列

主要分为两个类型的消息队列，分别支持广播模式和分组消费模式。

方案一

用来做消息队列并不是一个很好的选择，主要有两个原因。

第一个原因是XREAD读消息的特点是所有客户端共享队列中的所有消息，是典型的广播模式和发布/订阅模式类型，在数据安全方面还要比发布/订阅模式要好一些，后来加入的客户端还是可以通过start的设置来获取到历史的消息。

共享队列消息还不能发挥多客户端处理的优势，做即时通讯的队列服务还不错，如果是为了负载处理服务就完全没有效果了。

第二个原因是XREAD是读取消息队列的数据并没有确认操作，需要借助XDEL删除队列消息或者在创建队列是使用MAXLEN来限制队列大小。

XDEL操作不是作者推荐的方式，是因为stream队列是使用基数树来存储索引的，当删除一个消息时，并不是真正的树中将结点删除，而是标记为删除，只有点宏结点中所有结点都被删除了才会销毁该结点。

当频繁的对队列进行删除操作但没有全部删除时，会出现队列占用的内存空间超出你的预期。

而限制消息队列长度的方式也有弊端，首先客户端是共享数据的，不能提高处理能力，当生产者产生消息的速度远大于消费速度时会丢失消息，如果把限制值放的很大是又会出现过多的资源浪费。

综上所述，使用XADD+XREAD+XDEL实现的有序消息队列是不推荐的，还不如直接上List实现。

方案二

方案二考虑就完善多了，下面会围绕以下6个特点介绍：

使用XADD命令可以向队列中添加消息，当被添加的key不存在时创建新的队列。

XADD拥有子命令MAXLEN，正如上面提到的它可以设置队列的消息上限，你还可以通过~来设置一个不精确的上限，例如MAXLEN~ 1000。

XADD命令的返回值是消息ID，不管你是自定义的ID或者通过*使用默认的ID，ID都必须满足单向递增且唯一的特点。

默认ID是毫秒级的时间戳+当前时间的消息序号组成的，两个数字都是64的，基本上是没有机会重复，还有大型分布式项目都是靠Redis来生唯一ID的，所以最好还是使用默认的ID生成规则。

使用XGROUP来对队列设置分组，把客户端与分组进行关联。

使用XREADGROUP来获取分组后的消息，读消息时会告知分组名称和当前客户端名称，为后面的消息监控和消息流转做支持。

XREADGROUP与XREAD的操作基本上是一致的，都有COUNT子命令来获取返回消息数量；BLOCK选择阻塞时长，单位毫秒，时长为0时永远阻塞，默认不是要BLOCK是非阻塞模式；STREAMS限定接收的消息ID。

使用XACK确认消息处理，使用XREADGROUP读取的消息，服务器将会记住消息传递的客户端名称，消息会被存储在消费者组内的待处理条目列表（PEL）中，即已送达但尚未确认的消息ID列表。

可以使用XPENDING来查看未确认的消息ID列表，可以查看知道的客户端的PEL，返回值包括消息ID、客户端名称（消息的所有者）、消息传递给客户端到现在的毫秒数、该消息被传递的次数。

在消息被确认处理前消息的所有者是有可能宕机的，短时间内无法恢复的时候，我们不能让这个消息一直在所有者的PEL，需要使用XCLAIM命令把消息传递给新的客户端，让消息继续被处理。

使用XINFO命令来实现对服务器信息的监控。

方案二除了我们前面两天提到的优先队列不能实现以外，对于消息的确认机制、阻塞模式、处理中消息的监控、重复消息的问题都有了解决方案。

最终好不好用还需要来个测试项目感受下

今天就到这里了，退下

感觉用就点个关注、点个看一看噢

end