Redis Spring数据中是否支持XClaim /声明 - ReactiveRedisOperations.opsForStream()答案

【问题标题】：Is XClaim / claim supported in redis spring data - ReactiveRedisOperations.opsForStream()Redis Spring数据中是否支持XClaim /声明 - ReactiveRedisOperations.opsForStream()
【发布时间】：2020-09-19 16:46:06
【问题描述】：

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

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

【问题讨论】：

标签： redis spring-data spring-data-redis lettuce spring-data-redis-reactive

【解决方案1】：

如果没有 spring data redis，直接使用 lettuce 客户端库，我可以获取待处理的消息并声明如下消息

public Flux<PendingMessage> getPendingMessages(PollMessage pollMessage, String queueName) {
    Predicate<PendingMessage> poisonMessage = pendingMessage -> (pendingMessage.getTotalDeliveryCount()<=maxRetries);
    Predicate<PendingMessage> nackMessage = pendingMessage -> (pendingMessage.getElapsedTimeSinceLastDelivery().compareTo(Duration.ofMillis(ackTimeout)) > 0 );

    return statefulRedisClusterConnection.reactive()
        .xpending(queueName, pollMessage.getConsumerGroupName(), Range.unbounded(), Limit.from(1000))
        .collectList()
        .map((it) -> ((PendingMessages)PENDING_MESSAGES_CONVERTER
                .apply(it, pollMessage.getConsumerGroupName()))
                .withinRange(org.springframework.data.domain.Range.unbounded()))
            .flatMapMany(Flux::fromIterable)
            .filter(nackMessage)
            .filter(poisonMessage)
            .limitRequest(pollMessage.getBatchSize());
}

为了获取消息，我再次使用了生菜库中提供的 api

public Flux<StreamMessage<String, String>> claimMessage(PendingMessage pendingMessage, String queueName, String groupName, String serviceName) {
    return statefulRedisClusterConnection.reactive()
            .xclaim(queueName, Consumer.from(groupName, serviceName), 0, pendingMessage.getIdAsString());
}

目前，通过 spring-data 从 redis 获取待处理消息存在问题，因此我直接使用 lettuce 库来获取待处理消息并声明它。

https://jira.spring.io/browse/DATAREDIS-1160

【讨论】：