如何用Redisson锁定并解锁反应堆通量流？

Question

我有一个通量流，该流从数据库中读取对象。
对于这些对象，我都有一个要运行的处理函数。
我希望处理函数在给定对象的ID上获取REDIS锁定后，并在处理后释放锁定后执行（如果处理功能会引发错误）。

创建这样的流中最简单的通量方法是什么？
这是我尝试使用转换尝试执行此操作的一些代码。
我可能可以通过锁定功能，该函数将作为afterlock.flatmap（func）附加，但我正在寻找可以避免的解决方案。

我希望这与该流的其余部分一样透明，而不需要锁定和解锁功能的单独附件，只有一个可以执行“ Lock-Process-unlock”的附件。

        private <T> Function<Flux<T>, Publisher<T>> withLock(Function<T, String> keyExtractor) {
          
            return flux -> {
                    Flux<T> afterLock = flux.flatMap(ev -> redis.getLock(keyExtractor.apply(ev)).lock(1000L, TimeUnit.MILLISECONDS).map(ret -> ev));
    
    // processing logic should be attached somewhere here
    
                afterLock
                        .flatMap(ret -> redis.getLock(keyExtractor.apply(ret)).unlock()
                                .thenReturn(ret)
                                .onErrorResume(e -> redis.getLock(keyExtractor.apply(ret)).unlock().thenReturn(ret)));
    
                return afterLock;
    
            };
        }

        Flux.just(someObjectFromDatabase)
                .transform(withLock(t -> t.id()))
                .flatMap(this::objectProcessor)

Answer 1

解决方案之一是使用单声道。使用允许使用异步操作来用于资源供应商，资源关闭和清理。

Mono.usingWhen(
   lockService.acquire(key),
   lock -> process(),
   lock -> lockService.release(lock)
);

在我们的情况下，我们将redis锁定到lockservice看起来如下

public interface ReactiveLockService {

    Mono<LockEntry> acquire(String key, Duration expireAfter);

    Mono<Void> release(LockEntry lock);

    interface LockEntry {
        String getKey();

        String getValue();
    }
}

Answer 2

感谢您的回答@Alex，与此同时，我能够提供类似的东西，在组织流和失败方面非常灵活（花了我一段时间来掩盖边缘案例...）
它可以用作调用stream.flatmap（with lock（...，处理器））

    public static <T> Function<T, Flux<T>> withLock(
            long tid, String lockPrefix, int lockTimeMs, Function<T, String> keyExtractor, Function<Mono<T>, Flux<T>> processor, RedissonReactiveClient redis) {

        // If Redis lock or unlock operations fail, that will on purpose propagate the error.
        // If processor throws an error, lock will be unlocked first before error propagation.

        // tid has to be unique for each local task, it's a virtual "thread id" so if it's used concurrently locks will not protect the code

        return flux -> {
            Function<T, RLockReactive> getLock = ev -> redis.getLock(lockPrefix + keyExtractor.apply(ev));
            RLockReactive lock = getLock.apply(flux);

            Supplier<Mono<T>> unlock = () -> lock.unlock(tid).then(Mono.<T>empty());
            Supplier<Mono<T>> doLock = () -> lock.lock(lockTimeMs, TimeUnit.MILLISECONDS, tid).then(Mono.<T>empty());

            // Careful not to call map/flatMap on redis.lock/redis.unlock which returns Void and so it won't trigger on empty stream...lol!
            return Flux.concat(
                    Mono.defer(doLock),
                    Flux.defer(() -> processor.apply(Mono.just(flux))
                            .onErrorResume(err -> unlock.get()
                                    .onErrorResume(unlockError -> {
                                        err.addSuppressed(unlockError);
                                        // Propagate original processor error, but note the unlock error as suppressed
                                        return Mono.error(err);
                                    })
                                    .then(Mono.error(err)))),
                    Mono.defer(unlock)
            );
        };