时间戳

首页 > 解决方案 > 通量。有没有办法重试最后一个元素？

java - 通量。有没有办法重试最后一个元素？

问题描述

Flux 是否允许在不将指针指向初始位置的情况下重试发生的异常的操作？我的意思是从“有问题的”元素。

例如：

Flux.fromArray(new Integer[]{1, 2, 3})
        .delayElements(Duration.ofSeconds(1))
        .doOnNext(i -> {
            System.out.println("i: " + i);
            if (i == 2) {
                System.out.println("2 found");
                throw new RuntimeException("2!!!!!!!1");
            }
        })
        .retry(2)
        .subscribe();

将有以下输出：

i: 1
i: 2
2 found
i: 1
i: 2
2 found
i: 1
i: 2
2 found

当我希望看到这样的输出时：

i: 1
i: 2
2 found
i: 2
2 found
i: 2
2 found

PSskipUntil不是我要找的

标签： javafluxreactive

解决方案

不是我知道的，但我可能是错的。

但是，您可以自己为该特定步骤提供该逻辑。例如，但是创建您自己的 Consumer 并将重试逻辑包装在其中

public class RetryConsumer<T> implements Consumer<T> {

    private int                 retryCount;
    private Consumer<? super T> delegate;

    public RetryConsumer(int retryCount, Consumer<? super T> delegate) {
        this.retryCount = retryCount;
        this.delegate = delegate;
    }

    @Override
    public void accept(T value) {

        int currentAttempts = 0;
        while (currentAttempts < retryCount) {
            try {
                delegate.accept(value);
                break;
            } catch (Throwable e) {
                currentAttempts++;
                if (currentAttempts == retryCount) {
                    throw e;
                }
                //Still have some attempts left
            }
        }

    }
}

然后，您可以在 Flux 步骤中重用它，即

Flux.fromArray(new Integer[]{1, 2, 3})
    .doOnNext(new RetryConsumer<>(2 , i -> {
        System.out.println("i: " + i);
        if (i == 2) {
            System.out.println("2 found");
            throw new RuntimeException("Error");
        }
     }))
     .subscribe();

推荐阅读