Dlaczego przedmioty nie są zalecane w reaktywnych rozszerzeniach .NET?

Obecnie zajmuję się platformą Reactive Extensions dla .NET i przeglądam różne zasoby wprowadzające, które znalazłem (głównie http://www.introtorx.com )

Nasza aplikacja obejmuje szereg interfejsów sprzętowych, które wykrywają ramki sieciowe, będą to moje IObservables, a następnie mam różne komponenty, które będą zużywać te ramki lub przeprowadzać w jakiś sposób transformację danych i tworzyć nowy typ ramki. Będą też inne komponenty, które będą musiały wyświetlać na przykład co n-tą klatkę. Jestem przekonany, że Rx przyda się w naszej aplikacji, jednak zmagam się ze szczegółami implementacji interfejsu IObserver.

Większość (jeśli nie wszystkie) zasobów, które czytałem, mówi, że nie powinienem sam implementować interfejsu IObservable, ale używać jednej z dostarczonych funkcji lub klas. Z moich badań wynika, że ​​utworzenie a Subject<IBaseFrame>zapewniłoby mi to, czego potrzebuję, miałbym pojedynczy wątek, który odczytuje dane z interfejsu sprzętowego, a następnie wywołuje funkcję OnNext mojej Subject<IBaseFrame>instancji. Różne komponenty IObserver otrzymywałyby wówczas powiadomienia od tego podmiotu.

Moje zamieszanie wynika z porady podanej w dodatku do tego samouczka, gdzie jest napisane:

Unikaj używania typów przedmiotów. Rx jest faktycznie funkcjonalnym paradygmatem programowania. Używanie podmiotów oznacza, że ​​teraz zarządzamy stanem, który potencjalnie ulega mutacji. Radzenie sobie zarówno ze stanem mutacji, jak i programowaniem asynchronicznym w tym samym czasie jest bardzo trudne do osiągnięcia. Ponadto wiele operatorów (metod rozszerzających) zostało starannie napisanych, aby zapewnić utrzymanie prawidłowego i spójnego czasu życia subskrypcji i sekwencji; kiedy wprowadzasz przedmioty, możesz to przerwać. Przyszłe wydania mogą również spowodować znaczne obniżenie wydajności, jeśli jawnie użyjesz tematów.

Moja aplikacja jest dość krytyczna pod względem wydajności, oczywiście zamierzam przetestować wydajność korzystania z wzorców Rx, zanim przejdzie do kodu produkcyjnego; jednak martwię się, że robię coś, co jest sprzeczne z duchem frameworka Rx przy użyciu klasy Subject i że przyszła wersja frameworka zaszkodzi wydajności.

Czy jest lepszy sposób na robienie tego, co chcę? Sprzętowy wątek odpytywania będzie działał w sposób ciągły, niezależnie od tego, czy są jacyś obserwatorzy, czy nie (bufor sprzętowy będzie się archiwizował), więc jest to bardzo gorąca sekwencja. Muszę następnie przekazać odebrane ramki wielu obserwatorom.

Każda rada byłaby bardzo mile widziana.

Ok, jeśli zignorujemy moje dogmatyczne sposoby i razem zignorujemy „tematy są dobre / złe”. Spójrzmy na przestrzeń problemową.

Założę się, że masz albo 1 z 2 stylów systemu, do których musisz się przyznać.

1. System zgłasza zdarzenie lub oddzwonienie, gdy nadejdzie wiadomość
2. Musisz sondować system, aby sprawdzić, czy jest jakaś wiadomość do przetworzenia

W przypadku opcji 1, łatwo, po prostu zawijamy ją odpowiednią metodą FromEvent i gotowe. Do pubu!

W przypadku opcji 2 musimy teraz zastanowić się, jak to sondujemy i jak to zrobić skutecznie. Również kiedy otrzymamy wartość, w jaki sposób ją opublikujemy?

Wyobrażam sobie, że chciałbyś mieć dedykowany wątek do ankietowania. Nie chciałbyś, aby inny programista uderzał w ThreadPool / TaskPool i zostawił cię w sytuacji głodu w ThreadPool. Alternatywnie nie chcesz kłopotów z przełączaniem kontekstu (chyba). Więc załóżmy, że mamy własny wątek, prawdopodobnie będziemy mieć jakąś pętlę While / Sleep, w której będziemy sondować. Kiedy czek znajdzie jakieś wiadomości, publikujemy je. Cóż, wszystko to brzmi idealnie dla Observable.Create. Teraz prawdopodobnie nie możemy użyć pętli While, ponieważ nie pozwoli nam to kiedykolwiek zwrócić Disposable, aby umożliwić anulowanie. Na szczęście przeczytałeś całą książkę, więc jesteś biegły w planowaniu cyklicznym!

Wyobrażam sobie, że coś takiego mogłoby zadziałać. #Nie testowany

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

Powodem, dla którego naprawdę nie lubię przedmiotów, jest to, że zwykle jest to przypadek programisty, który nie ma jasnego projektu problemu. Zhakuj temat, włóż go tu i wszędzie, a potem pozwól biednemu deweloperowi wsparcia odgadnąć, co się dzieje w WTF. Kiedy używasz metod Create / Generate etc, lokalizujesz efekty w sekwencji. Możesz zobaczyć to wszystko w jednej metodzie i wiesz, że nikt inny nie rzuca okropnym efektem ubocznym. Jeśli widzę pola tematyczne, muszę teraz poszukać wszystkich miejsc w klasie, z której jest używany. Jeśli jakiś MFer ujawni go publicznie, wszystkie zakłady są wyłączone, kto wie, jak ta sekwencja jest używana! Async / Concurrency / Rx jest trudne. Nie musisz utrudniać tego, pozwalając efektom ubocznym i programowaniu przyczynowości jeszcze bardziej kręcić głową.

Generalnie powinieneś unikać używania Subject, jednak myślę, że w przypadku tego, co tutaj robisz, działają one całkiem dobrze. Zadałem podobne pytanie, kiedy natknąłem się na komunikat „unikaj tematów” w tutorialach Rx.

Cytując Dave'a Sextona (z Rxx)

„Podmioty są stanowymi składnikami Rx. Są przydatne, gdy trzeba utworzyć obserwowalne wydarzenie podobne do zdarzenia jako pole lub zmienna lokalna”.

Zwykle używam ich jako punktu wejścia do Rx. Więc jeśli mam jakiś kod, który musi mówić „coś się stało” (tak jak masz), użyłbym a Subjecti zadzwoniłbym OnNext. Następnie ujawnij to jako IObservablesubskrypcję dla innych (możesz użyć AsObservable()na swoim temacie, aby upewnić się, że nikt nie może przesyłać na temat i zepsuć rzeczy).

Możesz to również osiągnąć za pomocą zdarzenia .NET i użycia FromEventPattern, ale jeśli zamierzam zmienić to wydarzenie tylko w i IObservabletak, nie widzę korzyści z posiadania zdarzenia zamiast Subject(co może oznaczać, że brakuje mi coś tutaj)

Jednak to, czego powinieneś dość zdecydowanie unikać, to subskrybowanie a IObservablez a Subject, tj. Nie przekazuj a Subjectdo IObservable.Subscribemetody.

Często, gdy zarządzasz tematem, w rzeczywistości po prostu ponownie implementujesz funkcje już w Rx i prawdopodobnie nie jest to tak solidny, prosty i rozszerzalny sposób.

Kiedy próbujesz dostosować asynchroniczny przepływ danych do Rx (lub utworzyć asynchroniczny przepływ danych z takiego, który obecnie nie jest asynchroniczny), najczęstsze przypadki to zazwyczaj:

• Źródłem danych jest zdarzenie : jak mówi Lee, jest to najprostszy przypadek: użyj FromEvent i udaj się do pubu.

• Źródło danych pochodzi z operacji synchronicznej i chcesz odpytywanych aktualizacji (np. Usługi sieciowej lub połączenia z bazą danych): W tym przypadku możesz skorzystać z sugerowanego podejścia Lee lub w prostych przypadkach możesz użyć czegoś takiego jak Observable.Interval.Select(_ => <db fetch>). Możesz chcieć użyć DistinctUntilChanged (), aby zapobiec publikowaniu aktualizacji, gdy nic się nie zmieniło w danych źródłowych.

• Źródłem danych jest pewien rodzaj asynchronicznego interfejsu API, który wywołuje wywołanie zwrotne : w tym przypadku użyj Observable.Create, aby połączyć wywołanie zwrotne w celu wywołania OnNext / OnError / OnComplete na obserwatorze.

• Źródłem danych jest wywołanie, które blokuje dostęp do nowych danych (np. Niektóre synchroniczne operacje odczytu z gniazda): w tym przypadku można użyć Observable.Create, aby opakować kod imperatywny, który czyta z gniazda i publikuje w Observer. kiedy dane są odczytywane. Może to być podobne do tego, co robisz z tematem.

Korzystanie z Observable.Create a tworzenie klasy, która zarządza Subject, jest równoważne użyciu słowa kluczowego yield vs tworzenie całej klasy, która implementuje IEnumerator. Oczywiście możesz napisać IEnumerator tak, aby był tak przejrzystym i tak dobrym obywatelem jak kod zysku, ale który z nich jest lepiej zamknięty i wygląda lepiej? To samo dotyczy Observable.Create vs Managment Subjects.

Observable.Create zapewnia czysty wzorzec dla leniwej konfiguracji i czystego porzucenia. Jak to osiągnąć za pomocą klasy opakowującej przedmiot? Potrzebujesz jakiejś metody Start ... skąd wiesz, kiedy ją nazwać? Czy po prostu zawsze go uruchamiasz, nawet gdy nikt nie słucha? A kiedy skończysz, jak sprawisz, że przestaniesz czytać z gniazda / sondować bazę danych itp.? Musisz mieć jakąś metodę Stop i nadal musisz mieć dostęp nie tylko do IObservable, który subskrybujesz, ale przede wszystkim do klasy, która utworzyła Subject.

Dzięki Observable.Create wszystko jest w jednym miejscu. Ciało Observable.Create nie jest uruchamiane, dopóki ktoś nie zasubskrybuje, więc jeśli nikt nie subskrybuje, nigdy nie używasz swojego zasobu. I Observable.Create zwraca Disposable, które mogą czysto zamknąć twój zasób / wywołania zwrotne itp. - jest to wywoływane, gdy Observer anuluje subskrypcję. Czas życia zasobów, których używasz do generowania Observable, jest starannie powiązany z czasem życia samego Observable.

Cytowany tekst w bloku wyjaśnia, dlaczego nie powinieneś go używać Subject<T>, ale mówiąc prościej, łączysz funkcje obserwatora i obserwowalnego, jednocześnie wprowadzając jakiś stan pomiędzy nimi (czy to hermetyzujesz, czy rozszerzasz).

Tutaj masz kłopoty; obowiązki te powinny być oddzielne i odrębne od siebie.

To powiedziawszy, w twoim konkretnym przypadku radziłbym podzielić swoje obawy na mniejsze części.

Po pierwsze, masz gorący wątek i zawsze monitorujesz sprzęt pod kątem sygnałów, dla których można zgłaszać powiadomienia. Jak byś to zrobił normalnie? Wydarzenia . Więc zacznijmy od tego.

Zdefiniujmy, EventArgsże twoje zdarzenie zostanie uruchomione.

// 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's immutability.
    public IBaseFrame BaseFrame { get; private set; }
}

Teraz klasa, która uruchomi zdarzenie. Uwaga, może to być klasa statyczna (ponieważ zawsze mają nić monitorowania bufora sprzętowego), lub coś, co nazywamy na żądanie, który wyznaje , że . Będziesz musiał to odpowiednio zmodyfikować.

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

Masz teraz klasę, która ujawnia wydarzenie. Observables dobrze współpracują z wydarzeniami. Do tego stopnia, że ​​istnieje pierwszorzędna obsługa konwersji strumieni zdarzeń (pomyśl o strumieniu zdarzeń jako wielokrotnych uruchomieniach zdarzenia) na IObservable<T>implementacje, jeśli będziesz postępować zgodnie ze standardowym wzorcem zdarzenia, za pomocą metody statycznejFromEventPattern w Observableklasie .

Mając źródło twoich zdarzeń i FromEventPatternmetodę, możemy łatwo stworzyć IObservable<EventPattern<BaseFrameEventArgs>>( EventPattern<TEventArgs>klasa ucieleśnia to, co zobaczysz w zdarzeniu .NET, w szczególności instancję pochodną EventArgsi obiekt reprezentujący nadawcę), na przykład:

// 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's going to be hot
// as the events are hot.
IObservable<EventPattern<BaseFrameEventArgs>> observable = Observable.
    FromEventPattern<BaseFrameEventArgs>(
        h => source.HardwareEvent += h,
        h => source.HardwareEvent -= h);

Oczywiście chcesz IObservable<IBaseFrame>, ale to łatwe, używając Selectmetody rozszerzenia w Observableklasie, aby utworzyć projekcję (tak jak w LINQ, a wszystko to możemy zawrzeć w łatwej w użyciu metodzie):

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's 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);
}

Źle jest uogólniać, że Przedmioty nie nadają się do użycia jako interfejs publiczny. Chociaż jest z pewnością prawdą, że nie jest to sposób, w jaki powinno wyglądać podejście do programowania reaktywnego, jest to zdecydowanie dobra opcja ulepszania / refaktoryzacji dla klasycznego kodu.

Jeśli masz normalną właściwość z akcesorem zestawu publicznego i chcesz powiadamiać o zmianach, nic nie przemawia za zastąpieniem jej przez BehaviorSubject. INPC czy inne dodatkowe imprezy po prostu nie są takie czyste i to mnie osobiście męczy. W tym celu możesz i powinieneś używać BehaviorSubjects jako właściwości publicznych zamiast normalnych właściwości i porzucić INPC lub inne zdarzenia.

Dodatkowo interfejs Subject sprawia, że ​​użytkownicy twojego interfejsu są bardziej świadomi funkcjonalności twoich właściwości i są bardziej skłonni do subskrypcji, zamiast po prostu uzyskać wartość.

Najlepiej jest go używać, jeśli chcesz, aby inni słuchali / subskrybowali zmiany właściwości.