spring-boot - Reactive Redis 不会持续发布对 Flux 的更改

Question

我正在尝试在我的 redis 订购列表上获取实时更新，但没有成功。看起来它获取了所有项目并仅在最后一个项目上结束。我希望客户能够在我的订购列表中不断获取新订单的更新。我错过了什么？

这是我的代码:

@RestController
class LiveOrderController {

    @Autowired
    lateinit var redisOperations: ReactiveRedisOperations<String, LiveOrder>

    @GetMapping(produces = [MediaType.TEXT_EVENT_STREAM_VALUE], value = "/orders")
    fun getLiveOrders(): Flux<LiveOrder> {
        val zops = redisOperations?.opsForZSet()
        return zops?.rangeByScore("orders", Range.unbounded())
    }
}

Answer 1

Redis中没有这样的功能。首先，排序集的响应式(Reactive)检索只是获取快照，但您的调用是以响应式(Reactive)方式进行的。因此您需要订阅。

如果您选择加入 keyspace notifications像这样(K - 启用键空间通知，z - 包括 zset 命令):

config set notify-keyspace-events Kz

并在您的服务中订阅它们，如下所示:

  ReactiveRedisMessageListenerContainer reactiveRedisMessages;
  // ...
  reactiveRedisMessages.receive(new PatternTopic("__keyspace@0__:orders"))
      .map(m -> {
        System.out.println(m);
        return m;
      })
      <further processing>

您会看到如下消息:PatternMessage{channel=__keyspace@0__:orders, pattern=__keyspace@0__:orders, message=zadd} 。它会通知您已添加某些内容。你可以以某种方式对此使用react - 再次获取全套，或仅获取部分(头部/尾部)。您甚至可能还记得上一组，获取新一组并发送差异。

但我真正建议的是以某种方式重新构建流程以直接使用 Redis Pub/Sub 功能。例如:发布者服务而不是直接调用zadd会打电话eval ，它将发出 2 个命令:zadd orders 1 x和publish orders "1:x" (任何您想要的自定义消息，也许是 JSON)。

然后在您的代码中您将订阅您的自定义主题，如下所示:

return reactiveRedisMessages.receive(new PatternTopic("orders"))
      .map(LiveOrder::fromNotification);