RxJava thread-safety

Question

È questo il codice thread-safe?

Observable<String> observable = ... // some observable that calls 
            // onNext from a background thread 

observable 
    .scan(new ArrayList<String>(), (List<String> acc, String next) -> { 
    acc.add(next); 
    return acc; 
    }) 
    .subscribe(list -> { 
    // do somethind with sequence of lists 
    ... 
    }); 

Sono curioso perché ArrayList non è una struttura di dati thread-safe.

Answer 1

Come risposta rapida, in .NET (l'implementazione Rx originale) tutti i valori di una sequenza osservabile possono essere considerati sequenziali. Questo non preclude che sia multi-thread. Tuttavia, se si producono valori in modalità multi-thread, è possibile applicare la natura sequenziale cercando la funzione equivalente all'operatore .NET Synchronize() Rx.

Un'altra opzione è controllare l'implementazione di Scan nel codice sorgente RxJava, per verificare che faccia rispettare la natura sequenziale che si vorrebbe/prevedi di fornire sicurezza nella funzione accumulatore.

Answer 2

Se questo codice non è thread-safe, allora RxJava è rotto o la tua origine Observable è rotta - gli operatori non rientranti rientrano nel contratto Rx.