extensión reactiva OnNext

Question

Con Asunto RX, ¿es seguro para la ejecución de subprocesos llamar a OnNext() desde varios subprocesos?

Para que la secuencia se pueda generar desde varias fuentes.

Con fusión, ¿hacer lo mismo?

Answer 1

El contrato de Rx requiere que las notificaciones sean secuenciales, y es una necesidad lógica para varios operadores. Dicho esto, puede usar los métodos disponibles Synchronize para obtener este comportamiento.

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

Ahora puede hacer llamadas simultáneas al syncedSubject. Para un observador que debe ser sincronizado, también se puede utilizar:

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

prueba:

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

Answer 2

No, las secuencias están destinadas a ser secuenciales, por lo tanto, no se permiten notificaciones superpuestas. Puede utilizar los métodos de sincronización Synchronize para aplicar la sincronización adecuada. Los operadores como Merge toman un bloqueo para llamar al observador downstream a fin de garantizar una invocación en serie adecuada en On * callbacks.

Answer 3

Llamar a someSubject.OnNext() es tan seguro para subprocesos como someList.Add() - puede llamarlo desde> 1 hilo, pero no al mismo tiempo. Envuelva su OnNext en una declaración lock y será seguro.