RxJava中如何实现重试逻辑

在RxJava中，实现重试逻辑可以使用retry()和retryWhen()操作符。这两个操作符允许你在Observable发生错误时重新订阅，从而实现重试的功能。下面是它们的使用方法：

1. 使用retry()操作符：

retry()操作符接受一个整数参数，表示重试的次数。当Observable发生错误时，它会根据指定的次数重新订阅。这是一个简单的例子：

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;

public class RetryExample {
    public static void main(String[] args) throws InterruptedException {
        Observable.just("Hello")
                .map(s -> {
                    System.out.println("Processing: " + s);
                    throw new RuntimeException("Error occurred");
                })
                .retry(3) // 重试3次
                .subscribeOn(Schedulers.io())
                .subscribe(
                        s -> System.out.println("Success: " + s),
                        Throwable::printStackTrace
                );

        Thread.sleep(2000);
    }
}

在这个例子中，当Observable发生错误时，它会重试3次。但是，如果错误持续发生，那么在第三次重试之后，订阅将会失败，并打印堆栈跟踪。

2. 使用retryWhen()操作符：

retryWhen()操作符接受一个函数，该函数将错误Observable转换为另一个Observable。当转换后的Observable发出一个数据项时，原始Observable将会重新订阅。这使得你可以根据需要实现更复杂的重试逻辑。例如，你可以添加延迟、指数退避等。下面是一个例子：

import io.reactivex.Observable;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

import java.util.concurrent.TimeUnit;

public class RetryWhenExample {
    public static void main(String[] args) throws InterruptedException {
        Observable.just("Hello")
                .map(s -> {
                    System.out.println("Processing: " + s);
                    throw new RuntimeException("Error occurred");
                })
                .retryWhen(new Function<Observable<Throwable>, Observable<?>>() {
                    private int retryCount = 0;

                    @Override
                    public Observable<?> apply(Observable<Throwable> throwableObservable) throws Exception {
                        return throwableObservable.flatMap(error -> {
                            if (retryCount++ < 3) {
                                System.out.println("Retrying... (" + retryCount + ")");
                                return Observable.timer(1, TimeUnit.SECONDS); // 延迟1秒后重试
                            } else {
                                return Observable.error(error); // 超过重试次数，返回错误
                            }
                        });
                    }
                })
                .subscribeOn(Schedulers.io())
                .subscribe(
                        s -> System.out.println("Success: " + s),
                        Throwable::printStackTrace
                );

        Thread.sleep(10000);
    }
}

在这个例子中，当Observable发生错误时，它会尝试重试最多3次。每次重试之间会有1秒的延迟。如果在3次重试之后仍然发生错误，订阅将会失败，并打印堆栈跟踪。