Почему темы не рекомендуются в .NET Reactive Extensions?

Ответ 1

Ok, Если мы игнорируем мои догматические способы и игнорируем "предметы хорошие/плохие" все вместе. Давайте посмотрим на проблемное пространство.

Бьюсь об заклад, у вас есть 1 из 2 стилей системы, в которые вы должны пойти.

• Система вызывает событие или обратный вызов, когда приходит сообщение
• Вам необходимо опросить систему, чтобы узнать, есть ли какие-либо сообщения для обработки.

Для варианта 1, просто, мы просто завершим его соответствующим методом FromEvent, и мы закончили. В Паб!

Для варианта 2 нам нужно подумать о том, как мы проводим опрос этого и как это сделать эффективно. Также, когда мы получаем значение, как мы его публикуем?

Я бы предположил, что вам нужен выделенный поток для опроса. Вам не хотелось бы, чтобы какой-то другой кодер забил ThreadPool/TaskPool и оставил вас в ситуации голодания ThreadPool. В качестве альтернативы вам не нужны проблемы переключения контекста (я думаю). Поэтому предположим, что у нас есть собственный поток, у нас, вероятно, будет какой-то цикл While/Sleep, в который мы сидим, чтобы опросить. Когда чек обнаруживает некоторые сообщения, мы публикуем их. Ну все это звучит идеально для Observable.Create. Теперь мы, вероятно, не можем использовать цикл While, поскольку это не позволит нам когда-либо возвращать одноразовый, чтобы разрешить отмену. К счастью, вы прочитали всю книгу, так что подходите к рекурсивному планированию!

Я предполагаю, что это может сработать. #NotTested

public class MessageListener
{
    private readonly IObservable<IMessage> _messages;
    private readonly IScheduler _scheduler;

    public MessageListener()
    {
        _scheduler = new EventLoopScheduler();

        var messages = ListenToMessages()
                                    .SubscribeOn(_scheduler)
                                    .Publish();

        _messages = messages;
        messages.Connect();
    }

    public IObservable<IMessage> Messages
    {
        get {return _messages;}
    }

    private IObservable<IMessage> ListenToMessages()
    {
        return Observable.Create<IMessage>(o=>
        {
                return _scheduler.Schedule(recurse=>
                {
                    try
                    {           
                        var messages = GetMessages();
                        foreach (var msg in messages)
                        {
                            o.OnNext(msg);
                        }   
                        recurse();
                    }
                    catch (Exception ex)
                    {
                        o.OnError(ex);
                    }                   
                });
        });
    }

    private IEnumerable<IMessage> GetMessages()
    {
         //Do some work here that gets messages from a queue, 
         // file system, database or other system that cant push 
         // new data at us.
         // 
         //This may return an empty result when no new data is found.
    }
}

Причина, по которой мне действительно не нравятся темы, это, как правило, случай, когда разработчик не имеет четкого дизайна проблемы. Взломайте предмет, высуньте его здесь и везде, а затем позвольте бедной поддержке догадываться о том, что происходит в WTF. Когда вы используете методы Create/Generate и т.д., Вы локализуете эффекты в последовательности. Вы можете увидеть все это одним способом, и вы знаете, что никто не бросает неприятный побочный эффект. Если я вижу предметные поля, мне теперь нужно искать все места в классе, который он использует. Если какой-либо MFer раскрывает один из них публично, то все ставки отключены, кто знает, как эта последовательность используется! Async/ Concurrency/Rx является сложным. Вам не нужно делать это сложнее, позволяя побочным эффектам и программированию причинности еще больше откручивать голову.

Ответ 2

В общем, вам следует избегать использования Subject, однако для того, что вы здесь делаете, я думаю, что они работают неплохо. Я спросил аналогичный вопрос, когда я встретил сообщение "избегать вопросов" в учебниках Rx.

Процитировать Дейв Секстон (из Rxx)

"Субъекты представляют собой компоненты с состоянием Rx. Они полезны для вам необходимо создать событие, подобное наблюдаемому как поле или локальное переменная".

Я стараюсь использовать их как точку входа в Rx. Поэтому, если у меня есть код, который должен сказать "что-то случилось" (как и у вас), я бы использовал Subject и вызывал OnNext. Затем покажите, что в качестве IObservable для других, чтобы подписаться (вы можете использовать AsObservable() на свой предмет, чтобы убедиться, что никто не может бросить тему и испортить вещи).

Вы также можете добиться этого с помощью .NET-события и использовать FromEventPattern, но если я только собираюсь превратить событие в IObservable, я не вижу преимущества наличия события вместо a Subject (что может означать, что я здесь что-то не хватает)

Тем не менее, вы должны избегать довольно сильно подписываться на IObservable с Subject, то есть не передавать Subject в метод IObservable.Subscribe.

Ответ 3

Часто, когда вы управляете субъектом, вы на самом деле просто переопределяете функции уже в Rx и, вероятно, не настолько надежны, просты и расширяемы.

Когда вы пытаетесь адаптировать некоторый асинхронный поток данных в Rx (или создать асинхронный поток данных от одного, который в настоящее время не является асинхронным), наиболее распространенные случаи обычно:

• Источником данных является событие. Как говорит Ли, это самый простой случай: используйте FromEvent и отправляйтесь в паб.

• Источником данных является синхронная операция, и вы хотите получать опрошенные обновления (например, веб-сервис или вызов базы данных): в этом случае вы можете использовать предложенный Ли подход или просто вы можете использовать что-то вроде Observable.Interval.Select(_ => <db fetch>). Вы можете использовать DistinctUntilChanged() для предотвращения публикации обновлений, когда ничто не изменилось в исходных данных.

• Источником данных является какой-то асинхронный api, который вызывает ваш обратный вызов. В этом случае используйте Observable.Create для подключения вашего обратного вызова для вызова OnNext/OnError/OnComplete на наблюдатель.

• Источником данных является вызов, который блокируется до тех пор, пока не будут доступны новые данные (например, некоторые операции чтения синхронных сокетов): В этом случае вы можете использовать Observable.Create, чтобы обернуть императив код, который читается из сокета и публикуется в Observer.OnNext при чтении данных. Это может быть похоже на то, что вы делаете с субъектом.

Использование Observable.Create и создание класса, управляющего субъектом, довольно равнозначно использованию ключевого слова yield и создания целого класса, который реализует IEnumerator. Конечно, вы можете написать IEnumerator, чтобы быть чистым и хорошим гражданином, как код доходности, но который лучше инкапсулирован и выглядит более аккуратным. То же самое верно для Observable.Create vs управления объектами.

Observable.Create дает вам чистый образец для ленивой настройки и чистого разрыва. Как вы достигаете этого с классом, обертывающим тему? Вам нужен какой-то метод запуска... как вы знаете, когда его называть? Или вы всегда начинаете его, даже когда никто не слушает? И когда вы закончите, как вы его получите, чтобы прекратить чтение из сокета/опроса базы данных и т.д.? У вас должен быть какой-то метод Stop, и вы должны иметь доступ не только к IObservable, на который вы подписаны, но и классу, который создал тему в первую очередь.

С Observable.Create, все это завернуто в одном месте. Тело Observable.Create не запускается, пока кто-то не подписывается, поэтому, если никто не подписывается, вы никогда не используете свой ресурс. И Observable.Create возвращает одноразовый, который может полностью отключить ваш ресурс/обратные вызовы и т.д. - это вызывается, когда Observer отписывает. Время жизни ресурсов, которые вы используете для создания Observable, аккуратно привязано к времени жизни самого наблюдаемого.

Ответ 4

Цитата блочного текста в значительной степени объясняет, почему вы не должны использовать Subject<T>, но, проще говоря, вы комбинируете функции наблюдателя и наблюдаемые, в то же время вводя какое-то состояние между ними (независимо от того, являетесь ли вы инкапсулированием или расширением).

Здесь вы столкнулись с трудностями; эти обязанности должны быть отдельными и отличными друг от друга.

Тем не менее, в вашем конкретном случае я бы рекомендовал вам разделить свои проблемы на более мелкие части.

Во-первых, у вас есть ваш поток, который горячий, и всегда контролируйте аппаратное обеспечение сигналов для повышения уведомлений. Как бы вы это делали? События. Поэтому начнем с этого.

Определите EventArgs, что ваше событие будет срабатывать.

// The event args that has the information.
public class BaseFrameEventArgs : EventArgs
{
    public BaseFrameEventArgs(IBaseFrame baseFrame)
    {
        // Validate parameters.
        if (baseFrame == null) throw new ArgumentNullException("IBaseFrame");

        // Set values.
        BaseFrame = baseFrame;
    }

    // Poor man immutability.
    public IBaseFrame BaseFrame { get; private set; }
}

Теперь класс, который будет запускать событие. Обратите внимание, что это может быть статический класс (поскольку у вас всегда есть поток, контролирующий аппаратный буфер) или что-то, что вы вызываете по требованию, которое подписывается на это. Вы должны будете изменить это, если это необходимо.

public class BaseFrameMonitor
{
    // You want to make this access thread safe
    public event EventHandler<BaseFrameEventArgs> HardwareEvent;

    public BaseFrameMonitor()
    {
        // Create/subscribe to your thread that
        // drains hardware signals.
    }
}

Итак, теперь у вас есть класс, который предоставляет событие. Наблюдаемые хорошо работают с событиями. Настолько, что есть первоклассная поддержка для конвертации потоков событий (подумайте о потоке событий как о нескольких ошибках события) в IObservable<T> если вы следуете стандартным шаблонам событий, используя статический FromEventPattern метод на Observable класс.

С источником ваших событий и методом FromEventPattern мы можем легко создать IObservable<EventPattern<BaseFrameEventArgs>> (класс EventPattern<TEventArgs> воплощает то, что вы увидите в событии .NET, в частности, экземпляр, полученный из EventArgs и объект, представляющий отправителя), например:

// The event source.
// Or you might not need this if your class is static and exposes
// the event as a static event.
var source = new BaseFrameMonitor();

// Create the observable.  It going to be hot
// as the events are hot.
IObservable<EventPattern<BaseFrameEventArgs>> observable = Observable.
    FromEventPattern<BaseFrameEventArgs>(
        h => source.HardwareEvent += h,
        h => source.HardwareEvent -= h);

Конечно, вы хотите IObservable<IBaseFrame>, но это легко, используя Select метод расширения в классе Observable для создайте проекцию (точно так же, как в LINQ, и все это можно обернуть простым способом):

public IObservable<IBaseFrame> CreateHardwareObservable()
{
    // The event source.
    // Or you might not need this if your class is static and exposes
    // the event as a static event.
    var source = new BaseFrameMonitor();

    // Create the observable.  It going to be hot
    // as the events are hot.
    IObservable<EventPattern<BaseFrameEventArgs>> observable = Observable.
        FromEventPattern<BaseFrameEventArgs>(
            h => source.HardwareEvent += h,
            h => source.HardwareEvent -= h);

    // Return the observable, but projected.
    return observable.Select(i => i.EventArgs.BaseFrame);
}