郑州餐饮网站建设公司排名,郴州网上报名小学系统登录,自己做返利网站吗,免费的作文网站系列文章目录 文章目录 系列文章目录前言前言
前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住分享一下给大家。点击跳转到网站，这篇文章男女通用，看懂了就去分享给你的码吧。 Stream是一个只能追加内容的数据类型。也就是说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
