java - rxjava 当超时时跳过项目并继续下一个项目

Question

我有 3 件商品。当某个项目运行超时时，我想跳过该项目并继续下一个项目

我使用超时或throttleWithTimeout。没啥用

 Observable<String> just = ReplaySubject.just("1", "2", "3");
        just
                .doOnNext( a -> {
                    if("2".equals(a)){
                      TimeUnit.SECONDS.sleep(3);
                    }
                      System.out.println(a);
                })
                .timeout(2,TimeUnit.SECONDS)
                .onErrorResumeNext(o -> {
                    System.out.println("on error");
                }).subscribe();

控制台日志:

1
on error
2

如何实现超时项目跳过，并继续下一个项目。我期望输出是:

1
3

Answer 1

试试这个

Observable<String> just = ReplaySubject.just("1", "2", "3");
just
        .flatMap(aa -> Observable.just(aa)
                    .doOnNext(a -> {
                        System.out.println(a);
                        TimeUnit.SECONDS.sleep(3);
                    })
                    .timeout(2, TimeUnit.SECONDS, Observable.empty())
        )
        .subscribe();

每个项目都有超时，这就是为什么有 flatMap。使用 flatMap，您还可以使用 maxConcurrency 参数控制同时处理的元素数量。 Observable.empty表示超时时不做任何事情。