RxJava 背压（快速生产者慢消费者）

Question

我有执行方法，它在 io 线程上执行一些耗时的网络调用

例子

/**
 * network call
 * @param value
 * @return
 */
private Observable<Integer> execute(final int value) {
    return Observable.create(new Observable.OnSubscribe<Integer>() {
        @Override
        public void call(Subscriber<? super Integer> subscriber) {

            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("done " + value);
            subscriber.onNext(value);
            subscriber.onCompleted();
        }
    }).subscribeOn(Schedulers.io());
}

然后我有 "commands" 的列表，必须按顺序执行。（一个接一个）

示例（Observable.range(x,y) 表示命令列表）

public List<Integer> testObservableBackpressure(){
   return Observable.range(0,5).flatMap(new Func1<Integer, Observable<Integer>>() {
        @Override
        public Observable<Integer> call(Integer integer) {
            System.out.println("started " + integer);
            return exeute(integer);
        }
    }).toList().toBlocking().single();
}

这样输出就是

started 0
started 1
started 2
started 3
started 4
done 0
done 1
done 2
done 4
done 3

生产速度比消费速度快

我想要这样的结果

started 0
done 0
started 1
done 1
started 2
done 2
...

但是..

public List<Integer> testObservableBackpressure(){

    return Observable.create(new Observable.OnSubscribe<Integer>() {
        @Override
        public void call(final Subscriber<? super Integer> subscriber) {
            Observable.range(0,5).subscribe(new Subscriber<Integer>() {

                @Override
                public void onStart() {
                    request(1);
                }

                @Override
                public void onCompleted() {
                    subscriber.onCompleted();
                }

                @Override
                public void onError(Throwable e) {
                    subscriber.onError(e);
                }

                @Override
                public void onNext(Integer integer) {
                    System.out.println("started " + integer);
                    execute(integer).subscribe(new Action1<Integer>() {
                        @Override
                        public void call(Integer integer) {
                            subscriber.onNext(integer);
                            request(1);
                        }
                    });
                }
            });
        }
    }).toList().toBlocking().single();
}

这样的结果符合预期

started 0
done 0
started 1
done 1
started 2
done 2
started 3
done 3
started 4

我的问题是是否有另一种更优雅的方法来处理这个问题？

Answer 1

我运行你的代码得到的输出是：

started 0
started 1
started 2
started 3
started 4
done 1
done 3
done 4
done 2
done 0

请注意，"done" 消息顺序不对。这是因为您的代码基本上并行执行了对 execute 的每次调用。对于 Observable.range 发出的每个项目，您 flatMap 一个运行独立 IOScheduler 的 Observable。因此，每个项目都在单独的线程上并行处理，这使得项目不可能保持有序并正确交错。实现所需行为的一种选择是确保所有项目运行在同一个 IOScheduler 上（而不是每个项目单独）：

import rx.Observable;
import rx.Subscriber;
import rx.functions.Func1;
import rx.schedulers.Schedulers;

import java.util.List;

public class Test {
    private Observable<Integer> execute(final int value) {
        return Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {

                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("done " + value);
                subscriber.onNext(value);
                subscriber.onCompleted();
            }
        });
    }

    public List<Integer> testObservableBackpressure(){
        return Observable.range(0, 5).flatMap(new Func1<Integer, Observable<Integer>>() {
            @Override
            public Observable<Integer> call(Integer integer) {
                System.out.println("started " + integer);
                return execute(integer);
            }
        }).subscribeOn(Schedulers.io()).toList().toBlocking().single();
    }

    public static void main(String[] args) {
        new Test().testObservableBackpressure();
    }
}

请注意，唯一的区别是调用 subscribeOn 运算符的位置。此代码产生以下输出：

started 0
done 0
started 1
done 1
started 2
done 2
started 3
done 3
started 4
done 4

Answer 2

我不确定您是否需要任何特定的背压策略。只需使用 concatMap.

如果您使用 concatMap 而不是 flatMap，每个新输入值只会在 concatMap 发出的最后一个 Observable 完成时订阅。在幕后，concatMap 为此使用了 SerialSubscription。那应该给你你想要的顺序。

RxJava 背压（快速生产者慢消费者）

RxJava Backpressure (Fast producer slow consumer)

java

android

reactive-programming

rx-java