Redis中处理处理没有ACK确认的Stream

news2024/11/30 1:37:27

系列文章目录

文章目录

系列文章目录
前言

前言

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住分享一下给大家。点击跳转到网站，这篇文章男女通用，看懂了就去分享给你的码吧。
在这里插入图片描述

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- $] [SET I D k ey i d - 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