Unikanie nakładających OnNext wzywa w Rx przy użyciu SubscribeOn (Scheduler.TaskPool)

Question

mam jakiś kod za pomocą RX, wywoływana z wielu wątków, które wykonuje:

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

Chcę wartości mają być przetwarzane w tle, tak moja subskrypcja jest

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

i naprawdę nie obchodzi, które wątki obsługiwać wartości pochodzących z obserwowalnym, o ile praca jest umieścić w TaskPool i nie blokuje bieżącego wątku. Jednak korzystanie z "wartości" wewnątrz mojego delegata OnNext nie jest bezpieczne dla wątków. W chwili obecnej, jeśli wiele wartości przechodzi przez obserwowalne, uzyskuję nakładające się połączenia do mojej procedury obsługi OnNext.

Mogę tylko dodać blokadę do mojego delegata OnNext, ale to nie wydaje się być sposobem na robienie rzeczy. Jaki jest najlepszy sposób, aby upewnić się, że mam tylko jedno połączenie z moją obsługą OnNext w czasie, gdy mam wiele wątków wywołujących subject.OnNext(value);?

Answer 1

Od Using Subjects na MSDN

Domyślnie badani nie wykonują żadnej synchronizacji całej wątkach. [...] Jeśli jednak chcesz zsynchronizować połączenia wychodzące z obserwatorami za pomocą terminarza, możesz użyć metody Synchronizuj, aby to zrobić.

Powinieneś, jak powiedział Brandon w komentarzach, zsynchronizować temat i przekazać go swoim wątkom producenta. na przykład

var syncSubject = Subject.Synchronize(subject); 

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

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

Answer 2

Możesz spróbować zaimplementować własny serwer IObserver, który zastosuje blokadę w swojej metodzie OnNext. Po prostu prosty dekorator: zastosuj blokadę, zadzwoń do wewnętrznego OnNext, usuń blokadę.

Następnie można zaimplementować metodę rozszerzenia na IObservable, podobnie jak .AsThreadSafe().

Answer 3

Myślę, że szukasz metody rozszerzenia .Synchronize(). Aby uzyskać poprawę wydajności w najnowszej wersji (koniec 2011 r.), Zespół Rx złagodził założenia dotyczące sekwencyjnego charakteru obserwowalnych producentów sekwencji. Wygląda jednak na to, że przełamujesz te założenia (nie jest to zła rzecz), ale aby przywrócić grę Rx tak, jak oczekiwaliby użytkownicy, powinieneś Zsynchronizować sekwencję, aby upewnić się, że jest sekwencyjna ponownie.

Answer 4

Tutaj nieco więcej wyjaśnień, dlaczego używać Synchronize (drugi akapit). Z drugiej strony Synchronizacja może wziąć udział w impasie, jeśli aktywnie używasz blokowania w kodzie, przynajmniej byłem świadkiem takiej sytuacji.