如何计算 rx 中的处理时间

Question

对于以下流程，我想知道如何计算处理 forEach(...) 中所有数据所需的时间。

Observable
  .from(1,2,3)
  .flatMap(it - {})
  .toBlocking()
  .forEarch(it -> {//some paring logic here})

编辑

阅读本教程：Leaving the Monad 后，我觉得简单的解决方案是执行以下操作。如果我错过了什么，请告诉我

List items = Observable
      .from(1,2,3)
      .flatMap(it - {})
      .toList();

long startTime = System.currentTimeMillis();

for(Object it : items)
{
  //some parsing here
}

long processingTime = System.currentTimeMillis() - startTime 

Answer 1

一种选择是创建一个 Observable 来输出时间。您可以通过用以下方式包装计算来做到这一点Observable#using：

public class TimerExample {
    public static void main(String[] args) {
        final PublishSubject<Long> timings = PublishSubject.create();

        final Observable<List<Integer>> list = Observable
                .just(1, 2, 3)
                .flatMap(TimerExample::longRunningComputation)
                .toList();

        final Observable<List<Integer>> timed
                = Observable.using(() -> new Timer(timings), (t) -> list, Timer::time);

        timings.subscribe(time -> System.out.println("Time: " + time + "ms"));

        List<Integer> ints = timed.toBlocking().last();
        System.out.println("ints: " + Joiner.on(", ").join(ints));

        ints = timed.toBlocking().last();
        System.out.println("ints: " + Joiner.on(", ").join(ints));
    }

    private static Observable<Integer> longRunningComputation(Integer i) {
        return Observable.timer(1, TimeUnit.SECONDS).map(ignored -> i);
    }

    public static class Timer {
        private final long startTime;
        private final Observer<Long> timings;

        public Timer(Observer<Long> timings) {
            this.startTime = System.currentTimeMillis();
            this.timings = timings;
        }

        public void time() {
            timings.onNext(System.currentTimeMillis() - startTime);
        }
    }
}

在这种情况下，时间会打印到控制台，但您可以随意使用它们：

Time: 1089ms
ints: 2, 1, 3
Time: 1003ms
ints: 1, 3, 2