RxJava thread-safety

Question

Этот код безопасен?

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 
    ... 
    }); 

Мне любопытно, что ArrayList не является потокобезопасной структурой данных.

Answer 1

Как быстрый ответ, в .NET (исходная реализация Rx) все значения из наблюдаемой последовательности можно считать последовательными. Это не исключает многопоточности. Однако, если вы создаете значения в многопоточном режиме, вы можете потребовать последовательный характер, ища эквивалентную функцию для оператора Rx .NET Synchronize().

Другой вариант - проверить реализацию Scan в исходном коде RxJava, чтобы подтвердить, что он обеспечивает последовательный характер, который вы хотите/ожидаете, чтобы обеспечить безопасность в вашей функции аккумулятора.

Answer 2

Если этот код не является потокобезопасным, то либо RxJava сломан, либо поврежден источник Observable - операторы, не являющиеся второстепенными, являются частью контракта Rx.