问题标签 [spring-data-redis-reactive]

我试图在反应式编程中获得从 redis 读取的执行时间，在查找文档时我能够看到该elapsed()方法将执行相同的操作并实现如下代码。

输出：

我预计每个缓存请求所花费的时间将小于 5 毫秒，就像第一个和第二个请求一样，但情况并非如此。是否elapsed()将当前获取时间添加到累积中？根据我的理解，从通量发出的每个项目都是独立的？

如果它在 Redis 中，我需要将一些数据推送到客户端，但客户端会每 5 秒重新连接到 SSE 端点。

后端代码：

客户端使用 JS 消费：

谁能指出我正确的方向？
任何意见，将不胜感激。

更新：我需要在 Redis 中存储数据一段时间（5-10 分钟）。更新：我在 MongoDB 上写了类似的代码，它工作正常。

我ReactiveRedisConnection用来配置与本地 redis 容器的连接。

但将来应用程序将托管在 Web 服务器上，而 redis 将托管在不同的服务器上。

是否有任何选项可以为请求设置超时？

我正在使用 Spring 数据 Redis 从 Redis 流中消费，使用反应式流接收器来侦听消费者组的工作，但观察到 Flux 流有时会过早关闭并且不再收听新消息并且通量终止过早地。

代码

从流中读取消息一段时间后，获取“消费者流终止”的日志

版本：2.2.0.RELEASE

这是一个错误还是我错过了什么，有人可以帮忙吗？

更新

看起来 redis 命令正在超时，因为我收到了 RedisCommandTimeoutException，有没有办法在此类错误上重试流式处理而不是取消它。还发现它发生在 XREADGROUP 操作中，尽管通过 nodejs redis-cli 运行时发出相同的命令工作正常？

我一直在尝试使用 将Jackson2JsonRedisSerializer自定义实例序列化为 Redis 存储中的哈希值。似乎即使我已经正确创建了模板，也没有创建散列。

只是我与 Kotlin 一起使用的spring-data-reactive-redis注释spring-webflux。

这是使用此模板添加数据的示例。

谁能帮我解释为什么 Spring 不使用我创建的模板。

我正在学习使用 spring webflux，作为其中的一部分，我开发了一个使用 Redis 来保存和检索数据的应用程序。但我面临的问题是，当请求尝试连接到 redis 时，我收到以下错误：

我已经对这个问题进行了足够多的搜索，并试图找到任何有用的东西。

这是我的 Redis 配置类：

这是我的 redis 查找类，它实际上从 redis 获取和保存数据

控制器 ：

pom.xml

请帮我确定问题。提前致谢 ！

我是反应式编程的新手。我需要连接到 Redis 来保存和获取一些数据。redis 实例存在于云中。我正在使用生菜连接工厂建立连接。

与redis建立连接时，请求失败。这是我的 Redis 配置类：

这是我的查找服务类，它在请求期间实际上与 redis 通信

堆栈跟踪 ：

2020-03-26T16:27:54.513+0000 [APP/PROC/WEB/0] OUT org.springframework.data.redis.RedisConnectionFailureException: Unable to connect to Redis; nested exception is io.lettuce.core.RedisConnectionException: Unable to connect to 10.11.241.101:36516 | at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory$SharedConnection.getNativeConnection(LettuceConnectionFactory.java:1199) | Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: | Error has been observed at the following site(s): | |_ checkpoint ? Handler com.sap.slh.tax.attributes.determination.springwebfluxdemo.controller.TaxLinesDeterminationController#saveTaxLines(RedisRepo) [DispatcherHandler] | |_ checkpoint ? HTTP POST "/tax/lines/save/" [ExceptionHandlingWebHandler] | Stack trace: | at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory$SharedConnection.getNativeConnection(LettuceConnectionFactory.java:1199) | at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory$SharedConnection.getConnection(LettuceConnectionFactory.java:1178) | at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory.getSharedReactiveConnection(LettuceConnectionFactory.java:952) | at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory.getReactiveConnection(LettuceConnectionFactory.java:429) | at org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory.getReactiveConnection(LettuceConnectionFactory.java:94) | at org.springframework.data.redis.core.ReactiveRedisTemplate.lambda$doInConnection$0(ReactiveRedisTemplate.java:198) | at reactor.core.publisher.MonoSupplier.call(MonoSupplier.java:85) | at reactor.core.publisher.FluxUsingWhen.subscribe(FluxUsingWhen.java:80) | at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:55) | at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:150) | at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1705) | at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:241) | at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73) | at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:203) | at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:203) | at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1705) | at reactor.core.publisher.MonoIgnoreThen$ThenAcceptInner.onNext(MonoIgnoreThen.java:296) | at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1705) | at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:144) | at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1705) | at reactor.core.publisher.MonoZip$ZipCoordinator.signal(MonoZip.java:247) | at reactor.core.publisher.MonoZip$ZipInner.onNext(MonoZip.java:329) | at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onNext(MonoPeekTerminal.java:173) | at reactor.core.publisher.FluxDefaultIfEmpty$DefaultIfEmptySubscriber.onNext(FluxDefaultIfEmpty.java:92) | at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:67) | at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73) | at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1705) | at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:144) | at reactor.core.publisher.FluxContextStart$ContextStartSubscriber.onNext(FluxContextStart.java:103) | at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:287) | at reactor.core.publisher.FluxFilterFuseable$FilterFuseableConditionalSubscriber.onNext(FluxFilterFuseable.java:330) | at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1705) | at reactor.core.publisher.MonoCollect$CollectSubscriber.onComplete(MonoCollect.java:160) | at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136) | at reactor.core.publisher.FluxPeek$PeekSubscriber.onComplete(FluxPeek.java:252) | at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:136) | at reactor.netty.channel.FluxReceive.terminateReceiver(FluxReceive.java:419) | at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:209) | at reactor.netty.channel.FluxReceive.onInboundComplete(FluxReceive.java:367) | at reactor.netty.channel.ChannelOperations.onInboundComplete(ChannelOperations.java:363) | at reactor.netty.http.server.HttpServerOperations.onInboundNext(HttpServerOperations.java:489) | at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:90) | at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:377)

任何建议或答案都会非常有帮助！提前致谢 ！

在spring-data-redis-reactive中，写操作返回 redis 执行结果，这使得链接操作符非常困难。以Redis In Action的第 1 章中的示例 reddit 为例。我尝试像这样重新实现：

如您所见，我使用它doOnNext来避免丢失返回的 id 值increment，并且subscribe()在每个中都有一个doOnNext来确保每个 redis 操作都被执行。但我不认为这是推荐的方式。我认为应用程序应该尽量避免subscribe()并主要关注链接流。

做许多redis写操作的正确方法是什么spring-data-redis-reactive？

为了使用 redis 流构建可靠的消息队列，我使用 spring-boot-starter-data-redis-reactive 和 lettuce 依赖项来处理来自 redis 流的消息。虽然我可以通过以ReactiveRedisOperations.opsForStream()消费者组形式提供的 api 添加、读取、确认和删除消息，但我找不到一个 api 来声明一条未确认的待处理消息，尽管它在this.reactiveRedisConnectionFactory .getReactiveConnection() .streamCommands() .xClaim(). 但我不想有一个样板代码来管理异常、序列化等。有没有办法使用ReactiveRedisOperations.opsForStream()

https://docs.spring.io/spring-data/redis/docs/current/api/org/springframework/data/redis/core/ReactiveStreamOperations.html

我使用 spring Data Redis Reactive 框架，spring Boot 版本是 2.3.0。

这是我的 Redis 配置：

我使用 hashoperations.put 方法设置了一个 LocalDateTime 对象，以在 Redis 中显示以下表单：

当我使用 Mono<LocalDateTime>Mono = hashoperations.get(key,hashKey) 获取值时，会发生以下异常：

但是当 LocalDateTime 是对象的属性时，就没有问题了。

我不知道如何解决它。谢谢你的所有答案。