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

Question

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

附： skipUntil 不是我要找的东西

Answer 1

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

但是，您可以自己为该特定步骤提供该逻辑。例如，但是创建自己的 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();