複数スレッドからの先物 - onNextから観測可能

Question

Futuresのリストの結果からObservableをリアルタイムで生成したいと思います。

私はFuture.sequenceで実行している先物のリストを持っていて、完了するたびに通知するObservableで進捗を監視しています。私は基本的に次のようにしています：

def observeFuturesProgress(futs: List[Future[Int]]) : Observable[String] = { 
    Observable[String](observer => { 
     val loudFutures: List[Future[Int]] = futs.map(f => { 
      f onComplete { 
       case Success(a) => observer.onNext(s"just did $a more") 
       case Failure(e) => observer.onError(e) 
      } 
      f 
     }) 
     Future.sequence(loudFutures) onComplete { 
      case Success(_) => observer.onCompleted() 
      case Failure(e) => observer.onError(e) 
     } 
    }) 
    } 

これは私のテスト環境でうまくいきます。しかし、私はちょうど重複呼び出しがないことに注意しないで、onNextが異なるスレッドから呼び出されるべきではないことを読んだだけです。これを修正するための推奨される方法は何ですか？ Observablesの実際の使用方法の多くは、このように非同期コードからonNextを呼び出す必要がありますが、ドキュメントでも同様の例は見つかりません。

Answer 1

The Observable Contract

観測は（ない 並列で）シリアルオブザーバーに通知を発行する必要があります。彼らは 異なるスレッドからこれらの通知を発行するかもしれませんが、 通知の間に正式な先起こりの関係がなければなりません。

観察可能で、おそらく別のスレッドから、非同期にその観察者のメソッド を起動することが可能であるSerialize

を見てみましょう。これにより、 ObservableがObservableコントラクトに違反する可能性があります。 通知がOnNext 通知の前にOnCompletedまたはOnError通知を送信しようとするか、2つの 異なるスレッドから同時にOnNext通知を行う可能性があります。 シリアライズ演算子を適用することにより、このようなObservableを に正しく動作させ、同期させることができます。