Redis(消息队列Stream)

Stream是一个轻量级的消息队列。

Redis中Stream的作用是提供一种高效的消息传递机制，允许多个消费者并行地消费消息，并且不会重复消费已经处理过的消息。它可以用于实现分布式任务队列、日志收集、实时数据处理等场景。Redis中的Stream支持多个消费者组，每个消费者组可以独立消费消息，并且可以设置消费者组的消费位置，从而支持消息的多次消费和重放。同时，Redis的Stream还提供了一些高级特性，如持久化、自动截断、消息过期等，使得它更加适合在生产环境中使用。

Stream相关的命令都以X开头。

使用XADD添加返回的是消息ID，通过XLEN查看消息的数量。

通过使用XRANGE查询消息队列中的所有消息，包括消息ID，- +代表所有。

当然消息队列也是可以删除的，使用XDEL加消息ID就可以实现删除，这里就不做演示了。

删除消息也可以通过XTRIM来实现

XTRIM xbcl MAXLEN 0

上述代码代表删除所有消息

 

 在添加时可以自己指定消息ID，要保证这个ID是递增的，如果输入* Redis会保证消息ID是递增的 。

指定的ID第一个整数表示一个时间戳，第二个整数表示一个序列号。

可以使用XREAD来获取消息，COUNT后面的2代表获取的消息数量，BLOCK 1000代表如果没有消息就阻塞1000毫秒(返回nil)，geekhour 0代表从头开始获取(下标从0开始)。

如果希望获取从现在开始以后的最新消息就可以把最后的0改成$。

创建一个叫做group1的消费者组：

XGROUP GREATE geekhour group1 0

 

使用XINFO查询消费者组，XGROUP添加消费者。 

 

在连续添加了3个消费者后，查询消息发现group1组中有3个消费者 

使用XREADGROUP来读取消息，消息名字geekhour后的>表示从这个消息中读取最新消息

在Redis的XREADGROUP命令中，消息名称是由消费者组名和消息ID组成的。消息ID是一个由Redis自动生成的唯一标识符，它的格式为时间戳+序列号，例如"1631714987556-0"。在XREADGROUP命令中，消息名称的最后一个参数是">"符号，表示从当前时间开始获取最新的消息。如果想要获取指定消息ID之后的消息，可以将">"符号替换为指定的消息ID。例如，如果想要从消息ID为"1631714987556-0"之后获取消息，可以将消息名称的最后一个参数设置为"1631714987556-0"。

COUNT 2表示一次读取两条消息

输出就表明我们刚刚发布的消息被消费了。

为什么能解决订阅消息的问题：

Stream在Redis中是一种新的数据结构，用于解决发布/订阅模式中存在的一些问题。发布/订阅模式通常使用PUB/SUB命令来实现，但它存在一些限制和缺点，比如无法保证消息的持久性、消息的顺序性和消费者的灵活性等。

Stream通过引入消息ID、消费者组和消费者偏移量等概念，解决了这些问题。具体来说，Stream解决了以下几个问题：

1. 持久性：Stream中的消息会被持久化保存在内存中，即使没有订阅者，消息也不会丢失。这样就可以保证消息的持久性。

2. 顺序性：Stream中的消息是有序的，每个消息都有一个唯一的消息ID。消费者可以按照消息ID的顺序来消费消息，确保消息的顺序性。

3. 消费者灵活性：Stream引入了消费者组的概念，允许多个消费者共同消费一个消息流，而且每个消费者可以有自己的消费偏移量。这样就可以实现消费者的灵活性，比如可以动态地添加或移除消费者，而不会影响其他消费者。

4. 消息确认和重试：Stream中的消费者可以使用XACK命令来确认已经处理过的消息，同时可以使用XCLAIM命令来重新获取未确认的消息，从而实现消息的确认和重试。

总之，Stream通过引入消息ID、消费者组和消费者偏移量等概念，解决了发布/订阅模式中存在的一些问题，使得消息的处理更加灵活和可靠。