FWQ
详解RedisStream做消息队列
详解RedisStream做消息队列 收藏 亲爱的编程学习爱好者,如果你点开了这篇文章,说明你对《详解RedisStream做消息队列》很感兴趣。本篇文章就来给大家详细解析一下,主要介绍一下队列、消息、RedisStream,希望所有认真读完的童鞋们,都有实质性的提高。 List 众所周知redis数据结构中的list的lpush与rpop可以用于常规消息队列,从集合的最左端写入,最右端弹出消费。并且支持多个生产者与多个消费者并发拿数据,数据只能由一个消费者拿到。 但这个方案并不能保证消费者消费消息后是否成功处理的问题(服务挂掉或处理异常等),机制属于点对点模式不能做广播模式(发布/订阅模式) Pub/sub 于是redis提供了相应的发布订阅功能,为了解除点对点的强绑定模式引入了Channel管道。 当生产者向管道中发布消息,订阅了该管道的消费者能够同时接收到该消息,而且为了简化订阅多个管道需要显式关注多个名称提供了pattern能力。 通过名称匹配如果接收消息的频道wmyskxz.chat,consumer3也会收到消息。 但这个方案也有很大的诟病就是不会持久化,如果服务挂掉重启数据就全丢弃了,也没有提供ack机制,不保证数据可靠性,不管有没有消费成功发后既忘。 Stream stream的话结构很像kafka的设计思想,提供了consumer group和offset机制,结构上感觉跟kafka的topic差不多,只是没有对应partation副本机制,而是一个追加消息的链表结构。客户端调用XADD时候自动创建stream。每个消息都会持久化并存在唯一的id标识 Consumer Group 消费者组的概念跟kafka的消费者概念如出一辙,消费者既可以用XREAD命令进行独立消费,也可以多个消费者同时加入一个消费者组。一条消息只能由一个消费者组中的一个消费者消费。这样可以在分布式系统中保证消息的唯一性。 其实这个特性我后来仔细琢磨了一下当时自认为无懈可击的流式图表为了保证分布式系统消息唯一做了redis分布式锁。有点鸡肋,明明消费者组已经保证了数据的唯一性。只能说加锁可以压缩资源成本 last_delivered_id 用于标识消费者组消费在stream上消费位置的游标,每个消费者组都有一个stream内唯一的名称,消费者组不会自动创建,需要用XGROUP CREATE显式创建。 pending_ids 每个消费者内部都有一个状态变量。用来表示已经被客户端消费但没有ack的消费。目的是为了保证客户端至少消费了消息一次(atleastonce)。如果消费者收到了消息处理完了但是没有回复ack,就会导致列表不断增长,如果有很多消费组的话,那么这个列表占用的内存就会放大 curd xadd 追加消息 xdel 删除消息,这里的删除仅仅是设置了标志位,不影响消息总长度 xrange 获取消息列表,会自动过滤已经删除的消息 xlen 消息长度 del 删除Stream pending_ids如何避免消息丢失 在客户端消费者读取Stream消息时,Redis服务器将消息回复给客户端的过程中,客户端突然断开了连接,消息就丢失了。…