Избегание перекрытия OnNext-вызовов в Rx при использовании SubscribeOn (Scheduler.TaskPool)

У меня есть код с использованием Rx, вызываемый из нескольких потоков, который делает:

subject.OnNext(value); // where subject is Subject<T>

Я хочу, чтобы значения обрабатывались в фоновом режиме, поэтому моя подписка

subscription = subject.ObserveOn(Scheduler.TaskPool).Subscribe(value =>
{
    // use value
});

Мне все равно, какие потоки обрабатывают значения, выходящие из Observable, до тех пор, пока работа помещается в TaskPool и не блокирует текущий поток. Однако мое использование "значения" внутри моего делегата OnNext не является потокобезопасным. На данный момент, если через Observable проходит множество значений, я получаю перекрывающиеся вызовы для своего обработчика OnNext.

Я мог бы просто добавить блокировку в свой делегат OnNext, но это не похоже на Rx-способ делать что-то. Какой лучший способ убедиться, что у меня есть только один вызов для моего обработчика OnNext за один раз, когда у меня есть несколько потоков, вызывающих subject.OnNext(value);?

Ответ 1

Из Использование объектов в MSDN

По умолчанию субъекты не выполняют никакой синхронизации в потоки. [...] Если, однако, вы хотите синхронизировать исходящие вызовы с наблюдателями с помощью планировщика, вы можете использовать метод Synchronize для этого.

Так что, как говорит Брэндон в комментариях, синхронизируйте тему и передайте ее своим потокам. например.

var syncSubject = Subject.Synchronize(subject);

// syncSubject.OnNext(value) can be used from multiple threads

subscription = syncSubject.ObserveOn(TaskPoolScheduler.Default).Subscribe(value =>
{
    // use value
});

Ответ 2

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

Ответ 3

Здесь немного больше объяснений, почему использовать Synchronize (второй абзац). С другой стороны Synchronize может участвовать в тупике, если вы активно используете блокировку в своем коде, по крайней мере, я был свидетелем такой ситуации.

Ответ 4

Вы можете попробовать реализовать свой собственный IObserver, который будет применять блокировку в методе OnNext. Просто простой декоратор: примените блокировку, вызовите внутренний OnNext, удалите блокировку.

Затем вы можете реализовать метод расширения на IObservable, что-то вроде .AsThreadSafe().