沅江网站设计,凡客登录,app 开发 wordpress,计算机网站建设与开发Stream是一个轻量级的消息队列。
Redis中Stream的作用是提供一种高效的消息传递机制#xff0c;允许多个消费者并行地消费消息#xff0c;并且不会重复消费已经处理过的消息。它可以用于实现分布式任务队列、日志收集、实时数据处理等场景。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、消费者组和消费者偏移量等概念解决了发布/订阅模式中存在的一些问题使得消息的处理更加灵活和可靠。
