使用RxJava（ReactiveX）運行Observable需要多長時間？

Question

我在Scala中使用java ReactiveX（RxJava）Play Framework 2.5以異步方式與couchbase進行通信我想知道我的observable運行需要多長時間？我使用下面的代碼來定義我的observable。

def get(id: String) : Observable[Profile] = { 
    this.bucket 
    .async() 
    // can I have a start time here possibly using map? 
    .get(id) 
    .map[Profile](toProfile) 
    // can I have an end time here possibly using map? 
} 

我叫它使用以下

Thread.sleep(1000) 

val observable = get("myID") 

Thread.sleep(1000) 

// measure start time here 
println("observable: " + observable.toBlocking.first()) 
// measure end time here 

Thread.sleep(1000) 

如何衡量過了多長時間的觀察到的運行？

請多關照

弗朗西斯

Answer 1

你想開始在doOnSubscribe()塊的計時器，然後完成它在onTerminated()。

一個例子可能是這樣的：

long start; 
observable() 
    .doOnSubscribe(() -> start = System.nanoTime()) 
    .doOnTerminate(() -> System.out.println(System.nanoTime() - start)); 

或者你可以關注你的Netflix做RxNetty並返回起始時間爲流經鏈中的對象的一部分，並使用該結尾。