【Spring连载】使用Spring Data访问Redis（九）----Redis流 Streams

2024-02-04 12:20
文章标签 java 访问 使用 redis spring data 连载 streams

本文主要是介绍【Spring连载】使用Spring Data访问Redis（九）----Redis流 Streams，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

【Spring连载】使用Spring Data访问Redis（九）----Redis流 Streams

  • 一、追加Appending
  • 二、消费Consuming
    • 2.1 同步接收Synchronous reception
    • 2.2 通过消息监听器容器进行异步接收Asynchronous reception through Message Listener Containers
      • 2.2.1 命令式Imperative StreamMessageListenerContainer
      • 2.2.2 反应式Reactive StreamReceiver
    • 2.3 确认策略Acknowledge strategies
    • 2.4 读取偏移量策略ReadOffset strategies
  • 三、序列化Serialization
  • 四、对象映射Object Mapping
    • 4.1 简单值Simple Values
    • 4.2 复杂值Complex Values

Redis Streams以抽象的方法对日志数据结构进行建模。通常，日志是仅追加（append-only）的数据结构，并且从一开始就在随机位置或通过流式传输新消息来消费。
在 Redis参考文档中了解有关Redis Streams的更多信息。
Redis Streams大致可以分为两个功能领域：

  • 追加记录
  • 消费记录

尽管这种模式与Pub/Sub有相似之处，但主要区别在于消息的持久性以及消息的消费方式。
Pub/Sub依赖于瞬态消息的广播（即，如果你不听，你就会错过消息），而Redis Stream使用了一种持久的、仅追加的数据类型，它会保留消息，直到流被修剪。消费方面的另一个区别是Pub/Sub注册服务器端订阅。Redis将到达的消息推送到客户端，而Redis Streams需要活动轮询（active polling）。
org.springframework.data.redis.connection 和 org.springframework.data.redis.stream包为Redis Streams提供了核心功能。

一、追加Appending

要发送记录，你可以像使用其他操作一样，使用低级（low-level）RedisConnection或高级StreamOperations。这两个实体都提供add (xAdd)方法，该方法接受记录和目标流作为参数。RedisConnection需要原始数据（字节数组），而StreamOperations允许任意对象作为记录传入，如以下示例所示：

// append message through connection
RedisConnection con = …
byte[] stream = …
ByteRecord record = StreamRecords.rawBytes(…).withStreamKey(stream);
con.xAdd(record);// append message through RedisTemplate
RedisTemplate template = …
StringRecord record = StreamRecords.string(…).withStreamKey("my-stream");
template.opsForStream().add(record);

流记录携带一个Map，键值元组，作为它们的payload。将记录附加到流中会返回可作为进一步引用的RecordId。

二、消费Consuming

在消费端，你可以消费一个或多个流。Redis Streams提供读取命令，允许从已知流的任意位置（随机访问）消费流和从流的结束消费新的流记录。
在底层，RedisConnection提供了xRead和xReadGroup方法，它们分别映射Redis命令以在消费者组中进行各自读取。请注意，可以将多个流用作参数。
Redis中的订阅命令可能会被阻塞。也就是说，在连接(connection)上调用xRead会导致当前线程在开始等待消息时阻塞。只有当读取命令超时或收到消息时，线程才会被释放。
要消费流消息，可以在应用程序代码中轮询（poll）消息，也可以通过消息监听器容器使用两个异步接收中的一个（2.2章节），命令式或反应式。每次新记录到达时，容器都会通知应用程序代码。

2.1 同步接收Synchronous reception

虽然流消费通常与异步处理相关联，但也可以同步消费消息。重载的StreamOperations.read(…)方法提供了这个功能。在同步接收期间，调用线程可能会阻塞，直到消息可用为止。属性StreamReadOptions.block指定接收者在放弃等待消息之前应该等待多长时间。

// Read message through RedisTemplate
RedisTemplate template = …List<MapRecord<K, HK, HV>> messages = template.opsForStream().read(StreamReadOptions.empty().count(2),StreamOffset.latest("my-stream"));List<MapRecord<K, HK, HV>> messages = template.opsForStream().read(Consumer.from("my-group", "my-consumer"),StreamReadOptions.empty().count(2),StreamOffset.create("my-stream", ReadOffset.lastConsumed()))

2.2 通过消息监听器容器进行异步接收Asynchronous reception through Message Listener Containers

由于其阻塞性，低级别轮询（low-level polling）没有吸引力，因为它需要为每个消费者提供连接和线程管理。为了缓解这个问题，SpringData提供了消息监听器，它完成了所有繁重的工作。如果你熟悉EJB和JMS，你应该会发现这些概念很熟悉，因为它的设计尽可能接近Spring Framework中的支持及其消息驱动的POJO（MDP）。
Spring Data提供了两种针对所用编程模型量身定制的实现：

  • StreamMessageListenerContainer充当命令式编程模型的消息监听器容器。它用于消费Redis流中的记录，并驱动注入其中的StreamListener实例。
  • StreamReceiver提供了消息监听器的反应式变体。它用于将Redis流中的消息作为潜在的无限流消费，并通过Flux发出流消息。
    StreamMessageListenerContainer和StreamReceiver负责消息接收和分发到监听器中进行处理的所有线程。消息监听器容器/接收器是MDP和消息传递提供者之间的中介，负责注册接收消息、资源获取和释放、异常转换等。这使你作为应用程序开发人员能够编写与接收消息相关联的（可能复杂的）业务逻辑，并将Redis基础设施的公式化问题委托给框架。
    这两个容器都允许更改运行时配置，以便在应用程序运行时添加或删除订阅，而无需重新启动。此外，容器使用延迟订阅方法，仅在需要时使用RedisConnection。如果所有监听器都被取消订阅，它会自动执行清理，线程就会被释放。

2.2.1 命令式Imperative StreamMessageListenerContainer

以类似于EJB世界中的消息驱动Bean (MDB)的方式，流驱动POJO (SDP)充当流消息的接收者。SDP的一个限制是它必须实现org.springframework.data.redis.stream.StreamListener接口。还请注意，当POJO在多个线程上接收消息的情况下，要确保实现是线程安全的。

class ExampleStreamListener implements StreamListener<String, MapRecord<String, String, String>> {@Overridepublic void onMessage(MapRecord<String, String, String> message) {System.out.println("MessageId: " + message.getId());System.out.println("Stream: " + message.getStream());System.out.println("Body: " + message.getValue());}
}

StreamListener代表了一个函数式接口，因此可以使用Lambda形式重写实现:

message -> {System.out.println("MessageId: " + message.getId());System.out.println("Stream: " + message.getStream());System.out.println("Body: " + message.getValue());
};

一旦你实现了StreamListener，就该创建一个消息监听器容器并注册订阅了:

RedisConnectionFactory connectionFactory = …
StreamListener<String, MapRecord<String, String, String>> streamListener = …StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> containerOptions = StreamMessageListenerContainerOptions.builder().pollTimeout(Duration.ofMillis(100)).build();StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer.create(connectionFactory,containerOptions);Subscription subscription = container.receive(StreamOffset.fromStart("my-stream"), streamListener);

请参阅各种消息监听器容器的Javadoc，了解每种实现支持的特性的完整描述。

2.2.2 反应式Reactive StreamReceiver

流式数据源的反应式消费通常通过事件或消息的Flux发生。反应式接收器实现提供有StreamReceiver及其重载的receive(…)。与StreamMessageListenerContainer相比，反应式方法需要更少的基础设施资源，如线程，因为它利用了驱动程序提供的线程资源。接收流是StreamMessage的需求驱动（demand-driven）发布者：

Flux<MapRecord<String, String, String>> messages = …return messages.doOnNext(it -> {System.out.println("MessageId: " + message.getId());System.out.println("Stream: " + message.getStream());System.out.println("Body: " + message.getValue());
});

现在我们需要创建StreamReceiver并注册订阅来消费流消息:

ReactiveRedisConnectionFactory connectionFactory = …StreamReceiverOptions<String, MapRecord<String, String, String>> options = StreamReceiverOptions.builder().pollTimeout(Duration.ofMillis(100)).build();
StreamReceiver<String, MapRecord<String, String, String>> receiver = StreamReceiver.create(connectionFactory, options);Flux<MapRecord<String, String, String>> messages = receiver.receive(StreamOffset.fromStart("my-stream"));

请参阅各种消息监听器容器的Javadoc，了解每种实现支持的特性的完整描述。需求驱动的消费使用背压（backpressure）信号来激活和关闭轮询。如果需求得到满足，StreamReceiver订阅暂停轮询，直到订阅者发出进一步需求的信号。根据ReadOffset策略的不同，这可能导致消息被跳过。

2.3 确认策略Acknowledge strategies

当你通过Consumer Group读取消息时，服务器将记住给定的消息已传递并将其添加到Pending Entries List (PEL)中，它是已传递但尚未确认的消息列表。消息必须通过StreamOperations.acknowledge进行确认，才能从PEL中删除，如下面的代码段所示。

StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = ...container.receive(Consumer.from("my-group", "my-consumer"), ------1StreamOffset.create("my-stream", ReadOffset.lastConsumed()),msg -> {// ...redisTemplate.opsForStream().acknowledge("my-group", msg); ------2});1. 作为my-consumer从my-group中读取。接收到的消息不被确认。
2. 确认处理后的消息。

要在接收时自动确认消息，请使用receiveAutoAck而不是receive。

2.4 读取偏移量策略ReadOffset strategies

流读取操作接受读取偏移量规范，以消费给定偏移量的消息。ReadOffset表示读取偏移量规范。Redis支持3种偏移量变体，具体取决于你是单独使用流还是在消费者组中使用流：

  • ReadOffset.latest()–读取最新消息。
  • ReadOffset.from(…)–在特定消息Id之后读取。
  • ReadOffset.lastConsumed()–在上次消费的消息Id之后读取（仅限消费者组）。

在基于消息容器的消费上下文中，我们需要在消费消息时推进（或增加）读取偏移量。推进（advance）取决于请求的ReadOffset和消费模式（有没有消费者组）。以下表格说明了容器如何推进ReadOffset：
表1：推进ReadOffset

Read offsetStandaloneConsumer Group
Latest读取最新消息读取最新消息
特定Message Id使用最后看到的消息作为下一个MessageId使用最后看到的消息作为下一个MessageId
Last Consumed使用最后看到的消息作为下一个MessageId每个消费者组最后消费的消息

从特定的消息id和最后消费（last consumed）的消息中读取可以被认为是安全的操作，可以确保消费追加到流中的所有消息。使用最新消息进行读取可以跳过在轮询操作处于dead time状态时添加到流中的消息。轮询引入了一个dead time，在此期间消息可以在各个轮询命令之间到达。流消耗不是线性连续读取，而是分成重复的XREAD调用。

三、序列化Serialization

任何发送到流的记录都需要被序列化为二进制格式。由于流接近hash数据结构， stream key, field names 和 values使用RedisTemplate上配置的相应序列化器。
表2：流序列化

Stream Property序列化器描述
keykeySerializer用于Record#getStream()
fieldhashKeySerializer用于payload中map的每个key
valuehashValueSerializer用于payload中map的每个value

请务必查看正在使用的RedisSerializers，并注意如果你决定不使用任何序列化器，则需要确保这些值已经是二进制的。

四、对象映射Object Mapping

4.1 简单值Simple Values

StreamOperations允许通过ObjectRecord将简单的值直接追加到流中，而不必将那些值放入Map结构中。然后将该值分配给一个payload字段，并可以在读回该值时提取该值。

ObjectRecord<String, String> record = StreamRecords.newRecord().in("my-stream").ofObject("my-value");redisTemplate().opsForStream().add(record); ------1List<ObjectRecord<String, String>> records = redisTemplate().opsForStream().read(String.class, StreamOffset.fromStart("my-stream"));
1. 	XADD my-stream * "_class" "java.lang.String" "_raw" "my-value"

ObjectRecords和所有其他记录一样，都要经过相同的序列化过程，因此也可以使用返回MapRecord的非类型化读取操作获得Record。

4.2 复杂值Complex Values

向流添加复杂值可以通过以下三种方式完成：

  • 使用例如String JSON的表示转换为简单值。
  • 使用合适的RedisSerializer序列化该值。
  • 使用HashMapper将值转换为适合序列化的Map。

第一种变体是最直接的变体，但忽略了流结构提供的字段值功能，流中的值对其他消费者来说仍然是可读的。第二个选项与第一个选项具有相同的优点，但可能会导致非常特定的消费者限制，因为所有消费者都必须实现相同的序列化机制。HashMapper方法稍微复杂一点，它使用了stream哈希结构，但将源记录扁平化。只要选择了合适的序列化器组合，其他消费者仍然能够读取记录。
HashMappers将payload转换为具有特定类型的Map。请确保使用能够反序列化哈希的哈希键和哈希值序列化程序。

ObjectRecord<String, User> record = StreamRecords.newRecord().in("user-logon").ofObject(new User("night", "angel"));redisTemplate().opsForStream().add(record); -------1List<ObjectRecord<String, User>> records = redisTemplate().opsForStream().read(User.class, StreamOffset.fromStart("user-logon"));1. XADD user-logon * "_class" "com.example.User" "firstname" "night" "lastname" "angel"

StreamOperations默认使用ObjectHashMapper。在获得StreamOperations时，你可以提供适合你需求的HashMapper。

redisTemplate().opsForStream(new Jackson2HashMapper(true)).add(record); -----11. XADD user-logon * "firstname" "night" "@class" "com.example.User" "lastname" "angel"

StreamMessageListenerContainer可能不知道域类型上使用的任何@TypeAlias，因为这些需要通过MappingContext进行解析。确保用initialEntitySet初始化RedisMappingContext。

@Bean
RedisMappingContext redisMappingContext() {RedisMappingContext ctx = new RedisMappingContext();ctx.setInitialEntitySet(Collections.singleton(Person.class));return ctx;
}@Bean
RedisConverter redisConverter(RedisMappingContext mappingContext) {return new MappingRedisConverter(mappingContext);
}@Bean
ObjectHashMapper hashMapper(RedisConverter converter) {return new ObjectHashMapper(converter);
}@Bean
StreamMessageListenerContainer streamMessageListenerContainer(RedisConnectionFactory connectionFactory, ObjectHashMapper hashMapper) {StreamMessageListenerContainerOptions<String, ObjectRecord<String, Object>> options = StreamMessageListenerContainerOptions.builder().objectMapper(hashMapper).build();return StreamMessageListenerContainer.create(connectionFactory, options);
}

这篇关于【Spring连载】使用Spring Data访问Redis（九）----Redis流 Streams的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/677484。

相关文章

C语言中联合体union的使用

C语言中联合体union的使用

本文编辑整理自： http://bbs.chinaunix.net/forum.php?mod=viewthread&tid=179471 一、前言 “联合体”(union）与“结构体”(struct)有一些相似之处。但两者有本质上的不同。在结构体中,各成员有各自的内存空间, 一个结构变量的总长度是各成员长度之和。而在“联合”中,各成员共享一段内存空间, 一个联合变量
阅读更多...
Java五子棋之坐标校正

Java五子棋之坐标校正

上篇针对了Java项目中的解构思维，在这篇内容中我们不妨从整体项目中拆解拿出一个非常重要的五子棋逻辑实现：坐标校正，我们如何使漫无目的鼠标点击变得有序化和可控化呢？ 目录 一、从鼠标监听到获取坐标 1.MouseListener和MouseAdapter 2.mousePressed方法 二、坐标校正的具体实现方法 1.关于fillOval方法 2.坐标获取 3.坐标转换 4.坐
阅读更多...
Spring Cloud：构建分布式系统的利器

Spring Cloud：构建分布式系统的利器

引言 在当今的云计算和微服务架构时代，构建高效、可靠的分布式系统成为软件开发的重要任务。Spring Cloud 提供了一套完整的解决方案，帮助开发者快速构建分布式系统中的一些常见模式（例如配置管理、服务发现、断路器等）。本文将探讨 Spring Cloud 的定义、核心组件、应用场景以及未来的发展趋势。 什么是 Spring Cloud Spring Cloud 是一个基于 Spring
阅读更多...
Tolua使用笔记（上）

Tolua使用笔记（上）

目录   1.准备工作 2.运行例子 01.HelloWorld：在C#中，创建和销毁Lua虚拟机 和 简单调用。 02.ScriptsFromFile：在C#中，对一个lua文件的执行调用 03.CallLuaFunction：在C#中，对lua函数的操作 04.AccessingLuaVariables：在C#中，对lua变量的操作 05.LuaCoroutine：在Lua中，
阅读更多...
Javascript高级程序设计(第四版)--学习记录之变量、内存

Javascript高级程序设计(第四版)--学习记录之变量、内存

原始值与引用值 原始值：简单的数据即基础数据类型，按值访问。 引用值：由多个值构成的对象即复杂数据类型，按引用访问。 动态属性 对于引用值而言，可以随时添加、修改和删除其属性和方法。 let person = new Object();person.name = 'Jason';person.age = 42;console.log(person.name,person.age);//'J
阅读更多...
java8的新特性之一（Java Lambda表达式）

java8的新特性之一（Java Lambda表达式）

1：Java8的新特性 Lambda 表达式： 允许以更简洁的方式表示匿名函数（或称为闭包）。可以将Lambda表达式作为参数传递给方法或赋值给函数式接口类型的变量。 Stream API： 提供了一种处理集合数据的流式处理方式，支持函数式编程风格。 允许以声明性方式处理数据集合（如List、Set等）。提供了一系列操作，如map、filter、reduce等，以支持复杂的查询和转
阅读更多...
Vim使用基础篇

Vim使用基础篇

本文内容大部分来自 vimtutor，自带的教程的总结。在终端输入vimtutor 即可进入教程。 先总结一下，然后再分别介绍正常模式，插入模式，和可视模式三种模式下的命令。 目录 看完以后的汇总 1.正常模式（Normal模式） 1.移动光标 2.删除 3.【:】输入符 4.撤销 5.替换 6.重复命令【. ; ,】 7.复制粘贴 8.缩进 2.插入模式 INSERT
阅读更多...
Java面试八股之怎么通过Java程序判断JVM是32位还是64位

Java面试八股之怎么通过Java程序判断JVM是32位还是64位

怎么通过Java程序判断JVM是32位还是64位 可以通过Java程序内部检查系统属性来判断当前运行的JVM是32位还是64位。以下是一个简单的方法： public class JvmBitCheck {public static void main(String[] args) {String arch = System.getProperty("os.arch");String dataM
阅读更多...
详细分析Springmvc中的@ModelAttribute基本知识（附Demo）

详细分析Springmvc中的@ModelAttribute基本知识（附Demo）

目录 前言1. 注解用法1.1 方法参数1.2 方法1.3 类 2. 注解场景2.1 表单参数2.2 AJAX请求2.3 文件上传 3. 实战4. 总结 前言 将请求参数绑定到模型对象上，或者在请求处理之前添加模型属性 可以在方法参数、方法或者类上使用 一般适用这几种场景： 表单处理：通过 @ModelAttribute 将表单数据绑定到模型对象上预处理逻辑：在请求处理之前
阅读更多...
eclipse运行springboot项目，找不到主类

eclipse运行springboot项目，找不到主类

解决办法尝试了很多种，下载sts压缩包行不通。最后解决办法如图： help--->Eclipse Marketplace--->Popular--->找到Spring Tools 3---->Installed。
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2