本文主要是介绍Redis中处理处理没有ACK确认的Stream,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
系列文章目录
文章目录
- 系列文章目录
- 前言
前言
前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住分享一下给大家。点击跳转到网站,这篇文章男女通用,看懂了就去分享给你的码吧。
Stream是一个只能追加内容的数据类型。也就是说Stream这种数据类型,我们对他的添加操作,只能是向Stream的末尾追加内容,不能在头部或者中间插入内容。那追加的是什么内容呢?Stream中追加的内容其实就是一个或多个key-value pair。这些键值对不必遵循相同的结构。每一次追加的键值对都可以不同。例如第一次追加name=hello的键值对,第二次也可以变成追加desc=word的键值对。
关于Redis Stream的文章满天飞,这里不再重复,但是这些文章只是告诉你如何发消息,收消息,但是没有说如果我们的业务处理过程中出现了问题,这些没有处理的消息,怎么再次处理?
在之前消息队列时提过一个死信队列,Kafka 消费端消费重试和死信队列 https://www.javacui.com/tool/686.html ,那么Redis中,如何处理一些异常消息呢?
要理解下面的代码,首先要连接如下一些命令,为了解决这个问题,我也是查询了官方文档后,写出了下面处理代码。
XADD key ID field string [field string …]
将指定的流条目追加到指定key的流中。 如果key不存在,作为运行这个命令的副作用,将使用流的条目自动创建key。
一个条目是由一组键值对组成的,它基本上是一个小的字典。 键值对以用户给定的顺序存储,并且读取流的命令(如XRANGE 或者 XREAD) 可以保证按照通过XADD添加的顺序返回。
XADD是唯一可以向流添加数据的Redis命令,但是还有其他命令, 例如XDEL和XTRIM,他们能够从流中删除数据。
XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] STREAMS key [key …] ID [ID …]
XREADGROUP命令是XREAD命令的特殊版本,支持消费者组。
如果没有消费者组,仅使用XREAD,所有客户端都将获得所有到达流的条目。相反,如果使用带有XREADGROUP的消费者组,则可以创建不同的客户端组来消费到达给定流的不同的部分。例如,如果流获得新的条目A,B和C,并且有两个消费者通过消费者组读取流,其中一个客户端将会得到例如,消息A和C,另外一个客户端得到消息B,等等,以此类推。
XPENDING key group [start end count] [consumer]
通过消费者组从流中获取数据,而不是确认这些数据,具有创建待处理条目的效果。
XACK命令会立即从待处理条目列表(PEL)中移除待处理条目,因为一旦消息被成功处理,消费者组就不再需要跟踪它并记住消息的当前所有者。
XRANGE key start end [COUNT count]
此命令返回流中满足给定ID范围的条目。范围由最小和最大ID指定。所有ID在指定的两个ID之间或与其中一个ID相等(闭合区间)的条目将会被返回。
特殊ID:- 和 +,特殊ID-和+分别表示流中可能的最小ID和最大ID
XACK key group ID [ID …]
XACK命令用于从流的消费者组的待处理条目列表(简称PEL)中删除一条或多条消息。
一旦消费者成功地处理完一条消息,它应该调用XACK,这样这个消息就不会被再次处理, 且作为一个副作用,关于此消息的PEL条目也会被清除,从Redis服务器释放内存。
XGROUP [CREATE key groupname id-or- ] [ S E T I D k e y i d − o r − ] [SETID key id-or- ][SETIDkeyid−or−] [DESTROY key groupname] [DELCONSUMER key groupname consumername]
该命令用于管理流数据结构关联的消费者组。
这里使用Jedis进行编码测试,收到消息后没有手动ACK确认,用于演示处理此类数据,直接看代码
package com.example.springboot;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.StreamEntryID;
import redis.clients.jedis.params.XAddParams;
import redis.clients.jedis.params.XPendingParams;
import redis.clients.jedis.params.XReadGroupParams;
import
这篇关于Redis中处理处理没有ACK确认的Stream的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!