Реактивное расширение OnNext

С RX Subject, можно ли поточно-безопасным вызывать OnNext() из нескольких потоков?

Таким образом, последовательность может быть сгенерирована из нескольких источников.

Слиянием сделать то же самое?

Ответы

Ответ 1

Контракт Rx требует, чтобы уведомления были последовательными и являются логической необходимостью для нескольких операторов. Тем не менее, вы можете использовать доступные методы Synchronize, чтобы получить это поведение.

        var subject = new Subject<int>();
        var syncedSubject = Subject.Synchronize(subject);

Теперь вы можете совершать одновременные вызовы syncedSubject. Для наблюдателя, который должен быть синхронизирован, вы также можете использовать:

        var observer = Observer.Create<Unit>(...);
        var syncedObserver = Observer.Synchronize(observer);

Тест:

        Func<int, Action> onNext = i => () => syncedSubject.OnNext(i);
        Parallel.Invoke
        (
            onNext(1),
            onNext(2),
            onNext(3),
            onNext(4)
        );

Ответ 2

Нет, последовательности должны быть последовательными, поэтому не допускается перекрытие уведомлений. Вы можете использовать Синхронизировать методы расширения для обеспечения правильной синхронизации. Операторы, такие как Merge, блокируют вызов вызывающего потока наблюдателя, чтобы обеспечить правильный последовательный вызов в On * callbacks.

Ответ 3

Вызов someSubject.OnNext() как потокобезопасный как someList.Add() - вы можете вызывать его из > 1 потока, но не одновременно. Оберните OnNext в инструкции lock, и это будет безопасно.