初步了解Redis Streams以及如何在Java中使用它们

自今年年初以来，Redis Streams已进入Redis的unstable分支，并且第一个客户端始于采用Redis Streams API。 因此，这是一个绝佳的时机，可以从客户端角度看一下Redis Streams提供的功能以及如何使用它们。

免责声明：Redis Streams作为初稿提供，尚未成为稳定版本的一部分。 API可能会更改。

什么是Redis Stream？

Redis流是一种类似于日志/日志的数据结构，它按顺序表示事件的日志。 消息（事件）可以附加到流中。 然后可以以独立方式或通过在消费者组中阅读来使用这些消息。 使用者组是一个概念，其中可以将多个使用者（例如应用程序实例）分组为一个使用者组，其流偏移量（读取进度）保留在服务器端。 由于不需要在用户端保留流偏移，因此此概念简化了构建客户端的过程。

流消息由Redis在提交时生成的消息ID和表示为哈希（映射）的正文组成-基本上是一组键和值。
流本身由一个密钥标识，并保存零到许多流消息以及一些元数据，例如消费者组。

Redis Stream API

到目前为止，所有流命令都以X为前缀。 流允许使用添加，读取，自省和维护命令。 在下一部分中，您将看到最常见的命令：

  • XADD key * field1 value1 [field2 value2] [fieldN valueN] ：向Redis流添加（提交）一条消息。
  • XREAD [BLOCK timeout] [COUNT n] STREAMS key1 [keyN] offset1 [offsetN] ：从Redis流中读取消息。
  • XRANGE key from to [COUNT n] ：扫描（ XRANGE key from to [COUNT n] ）Redis流中的消息

此外，在使用使用者组时，还有其他命令在起作用：

  • XREADGROUP GROUP name consumer [BLOCK timeout] [COUNT n] [NOACK] STREAMS key1 [keyN] offset1 [offsetN] ：在使用者及其组的上下文中，从Redis流中读取消息。
  • XACK key group messageId1 [messageId2] [messageIdN] ：在使用者的上下文中读取后确认消息。
  • XPENDING key group [from to COUNT n] ：枚举未决（未确认的消息）。
  • XGROUP和子命令：用于创建和删除使用者组的API。

注意：为简洁起见，以上命令被截断了。 有关所有可能选项和组合的说明，请参见Redis Streams文档 。

使用Redis流

让我们来看一下如何通过redis-cli应用我们之前看到的命令来使用Redis Stream。 让我们向新流添加（并最初创建流）消息。 

127.0.0.1:6379> XADD my-stream * key value
1527062149743-0

我们正在使用XADD通过键值元组向流my-stream添加新消息。 注意* （星号）？ 这是用于控制ID生成的字段。 如果要由服务器生成消息ID（除非您是要复制的Redis服务器，否则在99.5％的用例中都是如此），请始终在其中添加* 。 Redis回复消息ID 1527062149743-0 。

现在，我们的信息流包含一条消息。 让我们用XREAD阅读它。 

127.0.0.1:6379>  XREAD COUNT 1 STREAMS my-stream 0
1) 1) "my-stream"2) 1) 1) 1527062149743-02) 1) "key"2) "value"

我们现在已经阅读了该消息，并沿着读取的内容检索了正文。 读取消息会将消息保留在流中。 我们可以使用XRANGE验证这XRANGE ： 

127.0.0.1:6379> XRANGE my-stream - +
1) 1) 1527068644230-02) 1) "key"2) "value"

发出具有相同流偏移量的后续读取将返回相同的消息。 您有不同的选择来避免此行为：

  1. 在客户端跟踪消息ID
  2. 阻止读取
  3. 从信息流中删除消息
  4. 限制流大小
  5. 使用消费者群体

让我们仔细看看这些选项。

MessageId追踪

每个读取操作都将返回消息ID和流消息。 如果您只有一个客户端（没有并发读取），则可以在应用程序中保留最新消息ID的引用，并在后续的读取调用中重用该消息。 让我们针对我们之前看到的1527068644230-0的消息ID进行此1527068644230-0 ： 

127.0.0.1:6379> XADD my-stream * key value
1527069672240-0
127.0.0.1:6379>  XREAD COUNT 1 STREAMS my-stream 1527068644230-0
1) 1) "my-stream"2) 1) 1) 1527069672240-02) 1) "key"2) "value"

我们使用1527068644230-0作为流偏移量并接收下一条添加的消息。 这种方法允许恢复读取较旧的（可能已经消耗的消息），但是需要在客户端进行一些协调以不读取重复的消息。

如果您不想跟踪消息ID，而仅对最新消息感兴趣，则可以使用阻塞读取。

阻止读取

通过XREAD读取允许以阻塞方式从流读取。 XREAD与BLPOP和BRPOP操作的行为类似，在BLPOP和BRPOP操作中您指定了超时，并且如果消息可用或读取超时，则调用将返回。 但是，Stream API允许更多选项。 对于此示例，我们需要两个单独的参与方：生产者和消费者。 如果您从头开始阅读，您将看到使用单个客户端执行的示例。 我们首先从消费者开始，否则产生的消息将到达流中而没有机会通知正在等待的消费者。

消费者

我们正在将XREAD与BLOCK 10000一起使用以等待10000毫秒（10秒）。 请注意，我们使用的符号流偏移量$指向流的开头。 

127.0.0.1:6379> XREAD COUNT 1 BLOCK 10000 STREAMS my-stream $

使用者现在被阻止，等待消息到达。

制片人 

127.0.0.1:6379> XADD my-stream * key value
1527070630698-0

Redis将消息写入流中。 现在，让我们切换回消费者。

消费者

消息写入我们的流之后，消费者收到一条消息并再次被解除阻止。 您可以开始处理该消息，并可能发出另一读。 

1) 1) "my-stream"2) 1) 1) 1527070630698-02) 1) "key"2) "value"
(1.88s)

使用流偏移量$发出另一个读取将再次等待到达该流的下一条消息。 但是，使用$会给我们提供一段时间，在此期间我们不会消耗其他消息。 为了避免这些漏洞，您应该跟踪上一次读取的消息ID，并将其重新用于下一个XREAD调用。
还要注意ist并发性。 我们已经看到一个单个消费者的例子。 如果增加消费者数量怎么办？

在这种情况下，例如，如果您有两个使用方发出阻塞读取，那么两个使用方都会收到同一条消息，这又使我们不得不承担协调读取的任务，因此流消息不会被多次处理。

从流中删除消息

可以从流中删除消息，但是不建议这样做。 我们还没有看到XDEL ，但是从名称上可以明显看出我们可以从流中删除消息： 

127.0.0.1:6379> XDEL my-stream 1527070789716-0
(integer) 1

该消息现在消失了。 不建议删除，因为操作成本很高：流使用带有宏节点的基数树。 删除是一种安全的操作，但是在与多个使用者一起使用一条消息时，您需要同步访问权限，因为删除不会阻止多次读取消息。

限制流大小

将消息追加到流时，可以指定最大流大小。 发出XADD命令时，使用MAXLEN选项会发生这种情况。 

127.0.0.1:6379> XADD my-stream MAXLEN 4 * key value
1527071269045-0

消息将添加到流中，并且将尽最大努力将流修剪到大小限制。 这也意味着较旧的消息将被修剪并且不再可读。

消费群体

解决重复消息处理的最后一种方法是利用使用者组。 消费者群体的想法是跟踪确认。 确认允许将消息标记为消费者确认。 XACK命令返回是否已确认该消息或先前的使用者是否已确认该消息。

要使用消费者组，我们需要首先创建一个消费者组。 请注意，自撰写本文时起，必须先存在一个流，然后才能创建消费者组。 这个问题可能将通过https://github.com/antirez/redis/issues/4824解决。

到目前为止，如果您遵循前面的示例，我们可以重用我们的流my-stream 。

我们正在创建一个名为my-group的使用者组，它仅对流my-stream有效。 请注意，最后一个参数是用于跟踪读取进度的流偏移量。 我们使用$指向流头。 

127.0.0.1:6379> XGROUP CREATE my-stream my-group $
OK

现在让我们向流中添加一条消息： 

127.0.0.1:6379> XADD my-stream * key value
1527072009813-0

并通过XREADGROUP发出非阻塞读取： 

127.0.0.1:6379> XREADGROUP GROUP my-group c1 COUNT 1 STREAMS my-stream >
1) 1) "my-stream"2) 1) 1) 1527072009813-02) 1) "key"2) "value"

XREADGROUP接受组名和使用者名来跟踪阅读进度。 另请注意流偏移量> 。 此符号流偏移量指向使用者组my-group读取的最新消息ID。
您可能已经注意到该组中有一个消费者名称。 消费者群体旨在跟踪消息传递并区分消费者。 如果您还记得上面的阻塞阅读示例，您已经看到两个使用者同时收到一条消息。 要更改（或保留）此行为，可以指定使用者名称：

  1. 具有相同使用者名称的读取可以多次接收相同的消息。
  2. 具有不同使用者名称的读取将阻止多次接收相同的消息。

根据您使用消息的方式，您可能想重新启动处理或使用多个客户端使用消息，而无需建立自己的同步机制。 Redis流允许您通过确认消息来做到这一点。 默认情况下， XREADGROUP确认消息，表明该消息已被处理并且可以被逐出。 您可以指定NOACK在阅读消息时不确认消息。 处理完消息后，确认消息发出XACK 。 根据返回的命令，您可以查看您是确认消息的对象还是其他客户端已经确认的消息。

现在让我们在这里暂停，不要再讨论恢复和更高级的主题。 Redis网站在https://redis.io/topics/streams-intro提供了有关Redis Streams的完整文档。

使用Java消耗Redis流

注意：在编写本文时，唯一支持Redis Streams的Java客户端是Lettuce预览版本5.1.0.M1。

Redis Streams带有新的服务器端API，该API也需要在客户端采用。 让我们使用Java客户端重播以上示例。

首先，我们需要一个客户端实例来准备连接。 我们将使用同步API。 但是，异步和反应式API也支持Redis Stream API。 

RedisClient client = RedisClient.create("redis://localhost");
StatefulRedisConnection<String, String> connection = client.connect();
RedisStreamCommands<String, String> streamCommands = connection.sync();

Lettuce引入了一个新的命令接口RedisStreamCommands ，该接口声明Redis Stream API方法及其各种类型（例如StreamOffset ， Consumer和命令参数对象）。

我们要做的第一件事是向流中添加新消息： 

Map<String, String> body =  Collections.singletonMap("key", "value");
String messageId = streamCommands.xadd("my-stream", body);

本示例使用UTF-8编码的字符串表示键和值。 主体本身作为Map传输，并发出命令XADD my-stream * key value 。

现在，让我们使用与XREAD COUNT 1 STREAMS my-stream 0相对应的命令从流中读取一条消息： 

List<StreamMessage<String, String>> messages = streamCommands.xread(XReadArgs.Builder.count(1), StreamOffset.from("my-stream", "0"));if(messages.size() == 1) { // a message was read} else { // no message was read}

所述xread(…)方法接受XReadArgs和StreamOffset并返回的列表StreamMessage<K, V>包含消息ID与主体一起对象。 现在可以处理消息了，随后的读取将包括最后一个messageId以读取新消息： 

StreamMessage<String, String> message = …;
List<StreamMessage<String, String>> messages = streamCommands.xread(XReadArgs.Builder.count(1), StreamOffset.from("my-stream", message.getId()));if(messages.size() == 1) { // a message was read} else { // no message was read}

阻塞读取需要将额外的持续时间传递到参数对象中。 添加BLOCK选项会将非阻塞调用（从Redis的角度来看）变成阻塞的调用： 

List<StreamMessage<String, String>> messages = streamCommands.xread(XReadArgs.Builder.count(1).block(Duration.ofSeconds(10)), StreamOffset.from("my-stream", "0"));

在最后一个示例中，让我们看一下消费者群体。 RedisStreamCommands提供了创建使用者的方法–截至撰写本文时，Redis中尚未实现删除使用者和使用者组的方法。 

streamCommands.xadd("my-stream", Collections.singletonMap("key", "value")); // add a message to create the stream data structurestreamCommands.xgroupCreate("my-stream", "my-group", "$"); // add a group pointing to the stream headList<StreamMessage<String, String>> messages = streamCommands.xreadgroup(Consumer.from("my-group", "c1"),StreamOffset.lastConsumed("my-stream"));

使用使用者组my-group和使用者c1从my-stream中读取消息。 使用者组和使用者名称是字节安全编码的，因此在使用ASCII或UTF-8字符串时区分大小写。

结论

这篇博客文章概述了Redis 5附带的Redis Streams的初步外观，以及如何与Lettuce Redis客户端一起使用Stream API。 该API尚未完全实现，因此我们应该期待更改。

翻译自: https://www.javacodegeeks.com/2018/05/a-first-look-at-redis-streams-and-how-to-use-them-with-java.html

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/346799.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

权限丢失_不要让任何“账户权限”问题困扰你！

权限丢失_不要让任何“账户权限”问题困扰你！

在Win10系统使用中时不时被“账户权限”问题困扰&#xff1f;别怕&#xff0c;有我呢~关闭更改系统相关设置时的通知不愿被安装软件、更改Windows设置时弹出的通知所打扰时&#xff0c;可以搜索【UAC】&#xff08;若没有搜索框&#xff0c;请使用【Win】【Q】快捷键调出&#…
阅读更多...
python函数返回的元组_python – 从函数返回一个单独的元组元组

python函数返回的元组_python – 从函数返回一个单独的元组元组

刚刚在Python中遇到了这样一些奇怪的事情&#xff0c;并且认为我会将其记录为一个问题&#xff0c;以防其他任何人试图用相同的无效搜索条件找到答案看起来像元组解包使得它不能返回长度为1的元组&#xff0c;如果您希望迭代返回值。虽然看起来似乎是欺骗。看到答案。>>&…
阅读更多...
java 反射操作字段_x86上的Java最终字段没有操作？

java 反射操作字段_x86上的Java最终字段没有操作？

java 反射操作字段我一直很乐于深入研究多线程编程的细节&#xff0c;并且尽管阅读了多年的CPU内存一致性模型&#xff0c;无等待和无锁算法&#xff0c;Java内存模型&#xff0c;实践中的Java并发性等知识&#xff0c;但我总是很喜欢。等等-我仍然会创建多线程编程错误。 总是…
阅读更多...
卡方分布分位数_卡方检验和精确概率法及两两比较

卡方分布分位数_卡方检验和精确概率法及两两比较

看过许多统计教程&#xff0c;这篇是我最推荐的介 绍数值变量如果服从正态分布&#xff0c;采用均数标准差进行统计描述&#xff0c;采用方差分析进行组间比较&#xff0c;如果组间差异有统计学意义&#xff0c;进一步采用LSD法(也可以是其它方法)进行两两比较。如果不服从正…
阅读更多...
arcgis渔网分割提取栅格图_【操作】ArcGIS中字段的合并、分割、提取

arcgis渔网分割提取栅格图_【操作】ArcGIS中字段的合并、分割、提取

由于种种原因&#xff0c;可能涉及到要对ArcGIS属性字段进行合并、分割、提取等操作&#xff0c;我们可以在Excel中进行运算&#xff0c;然后连接到ArcGIS中&#xff0c;这里主要介绍下通过属性表字段计算器的操作方法。目标① 合并“名称”、“代码”字段内容到新建字段“用地…
阅读更多...
ds排序--希尔排序_图解直接插入排序和希尔排序

ds排序--希尔排序_图解直接插入排序和希尔排序

前言这次我们介绍插入类排序中的 直接插入排序 和 希尔排序 。对于直接插入排序&#xff0c;虽然它的时间复杂度也是 O(n^2) &#xff0c;但是在元素 有序或近乎有序 的情况下&#xff0c;时间复杂度可以降为 O(n) &#xff0c;效率比 O(nlogn) 的算法还要高。然而对于大规模的…
阅读更多...
jsf 配置_JSF Tomcat配置示例

jsf 配置_JSF Tomcat配置示例

jsf 配置JavaServer Faces (JSF)是一个Web应用程序框架&#xff0c;旨在简化基于Web的用户界面的开发集成。 它用于开发和构建服务器端用户界面组件&#xff0c;并在Web应用程序中使用它们。 JSF技术基于Model-View-Controller (MVC)架构&#xff0c;并且通过在页面中使用可重用…
阅读更多...
tomcat8日志乱码_Tomcat输出日志乱码解决

tomcat8日志乱码_Tomcat输出日志乱码解决

请先检查一下是不是由于log4J没有指定编码方式所导致的.log4j.appender.File org.apache.log4j.DailyRollingFileAppender#重点是看一下有没有这个,如果没有则加上.#加上之后需要将之前的日志文件删除,否则有可能不会重新写入日志.log4j.appender.File.Encoding UTF-8log4j.a…
阅读更多...
使用Datastax Java驱动程序与Cassandra进行交互

使用Datastax Java驱动程序与Cassandra进行交互

今天&#xff0c;我这次返回了更多的Cassandra和Java集成&#xff0c;重点是使用Datastax Java驱动程序&#xff0c;而不是我已经写了很多文章的Spring Data Cassandra。 Spring Data实际上使用了Datastax驱动程序来与Cassandra进行交互&#xff0c;但是在其之上还附带了一些额…
阅读更多...
java forkjoin MySQL_Java并发fork-join框架

java forkjoin MySQL_Java并发fork-join框架

fork-join框架允许在几个工作进程中断某个任务&#xff0c;然后等待结果组合它们。 它在很大程度上利用了多处理器机器的生产能力。 以下是fork-join框架中使用的核心概念和对象。ForkFork是一个进程&#xff0c;其中任务将其分成可以并发执行的较小且独立的子任务。语法Sum le…
阅读更多...
无法加载可扩展计数器_多核可扩展计数器

无法加载可扩展计数器_多核可扩展计数器

无法加载可扩展计数器到处都需要计数器&#xff0c;例如&#xff0c;查找应用程序的关键KPI&#xff0c;应用程序的负载&#xff0c;服务的请求总数&#xff0c;用于查找应用程序吞吐量的一些KPI等。 由于所有这些需求&#xff0c;并发复杂性也增加了&#xff0c;这使这个问题…
阅读更多...
linux忘记mysql密码_linux下忘记mysql root密码解决办法 | 系统运维

linux忘记mysql密码_linux下忘记mysql root密码解决办法 | 系统运维

引言&#xff1a;在linux系统中&#xff0c;如果忘记了MySQL的root密码&#xff0c;有没有办法重新设置新密码呢&#xff1f;答案是肯定的&#xff0c;下面教大家一个比较简单的重置MySQL root密码的办法&#xff1a;1、编辑MySQL配置文件my.cnf系统运维www.osyunwei.com温馨提…
阅读更多...
Java 10中的本地类型推断，或者如果它像鸭子一样嘎嘎叫

Java 10中的本地类型推断，或者如果它像鸭子一样嘎嘎叫

最近&#xff0c;Oracle采用了一种新策略&#xff0c;即每六个月发布一次新的语言版本。 该策略假定只有每个第3版都将提供长期支持或LTS。 快速说明&#xff1a; 具有LTS的当前版本是Java 8&#xff1b; Java 9仅应在2018年3月之前获得支持&#xff0c;因此已经结束。 Java…
阅读更多...
mysql从dos界面_从DOS界面进入MYSQL数据库

mysql从dos界面_从DOS界面进入MYSQL数据库

新手学习数据库&#xff0c;从比较简单的mysql开始。从DOS界面进入mysql数据库其实跟配置java环境变量一样&#xff0c;要先设置环境变量(右击我的电脑-属性-高级-环境变量)。找到mysql的安装位置&#xff0c;把安装目录下“...mysql\bin\"这个地址放到环境变量Path值中&a…
阅读更多...
更快地重复访问Java的Java类名？

更快地重复访问Java的Java类名？

Claes Redestad在core-libs-dev邮件列表上发布了消息“ RRF&#xff1a;8187123 &#xff1a;&#xff08;反映&#xff09;Class&#xff03;getCanonicalName和Class&#xff03;getSimpleName是性能问题的一部分 ”&#xff0c;他在邮件列表中要求审查建议的更改“以启用缓存…
阅读更多...
mysql中cast函数_mysql中cast函数的使用 用于强制类型转换 (转载)

mysql中cast函数_mysql中cast函数的使用 用于强制类型转换 (转载)

例子&#xff1a;SELECTstr_to_date(concat(year(a.tb_time),date_format(a.tb_time,%m),01),%Y%m%d) tb_time,cast(a.category_id as char) category_id,a.category_name,sum(a.tendered_sum)/sum(a.winbid) tendered_sum,sum(a.winbid_sum)/sum(a.winbid) winbid_avgfrom( SE…
阅读更多...
dht11温湿度传感器_Arduino不调用库实现DHT11数据读取

dht11温湿度传感器_Arduino不调用库实现DHT11数据读取

DHT11概述DHT11数字温湿度传感器是一款含有已校准数字信号输出的温湿度复合传感器&#xff0c;它应用专用的数字模块采集技术和温湿度传感技术&#xff0c;确保产品具有极高的可靠性和卓越的长期稳定性。传感器包括一个电阻式感湿元件和一个NTC测温元件&#xff0c;并与一个高性…
阅读更多...
rete_RIP RETE时间来获得PHREAKY

rete_RIP RETE时间来获得PHREAKY

rete我刚刚完成了有关新规则算法PHREAK的高级文档&#xff0c;PHREAK是混合推理中的一个文字游戏。 它仍然有点粗糙和高水平&#xff0c;但希望仍然很有趣。 它建立在ReteOO之上&#xff0c;非常好阅读。 ReteOO算法 ReteOO是在3、4和5系列发行版中开发的。 它采用了RETE算法并…
阅读更多...
mysql 存储过程 插入记录_mysql 存储过程 插入记录

mysql 存储过程 插入记录_mysql 存储过程 插入记录

{"moduleinfo":{"card_count":[{"count_phone":1,"count":1}],"search_count":[{"count_phone":4,"count":4}]},"card":[{"des":"阿里云数据库专家保驾护航&#xff0c;为用户…
阅读更多...
Java技巧：创建监视友好的ExecutorService

Java技巧：创建监视友好的ExecutorService

在本文中&#xff0c;我们将扩展具有监视功能的ExecutorService实现。 这种监视功能将帮助我们在实时生产环境中测量多个池参数&#xff0c;即活动线程&#xff0c;工作队列大小等。 它还将使我们能够衡量任务执行时间&#xff0c;成功任务计数和失败任务计数。 监控库 至于监…
阅读更多...
最新文章

Copyright @ 2022~2023