Rx produce and consume on different threads
.net-4.0, c#, system.reactive
Solution
Understanding the difference between `ObserveOn` and `SubscribeOn` is key here. See - ObserveOn and SubscribeOn - where the work is being done for an in depth explanation of these.
Also, you absolutely don't want to use a `Thread.Sleep` in your Rx. Or anywhere. Ever. `Do` is almost as evil, but `Thead.Sleep` is almost always totally evil. Buffer has serveral overloads you want to use instead - these include a time based overload and an overload that accepts a count limit and a time-limit, returning a buffer when either of these are reached. A time-based buffering will introduce the necessary concurrency between producer and consumer - that is, deliver the buffer to it's subscriber on a separate thread from the producer.
Also see these questions and answers which have good discussions on keeping consumers responsive (in the context of WPF here, but the points are generally applicable).
- Process lots of small tasks and keep the UI responsive
- Buffer data from database cursor while keeping UI responsive
The last question above specifically uses the time-based buffer overload. As I said, using `Buffer` or `ObserveOn` in your call chain will allow you to add concurrency between producer and consumer. You still need to take care that the processing of a buffer is still fast enough that you don't get a queue building up on the buffer subscriber.
If queues do build up, you'll need to think about means of applying backpressure, dropping updates and/or conflating the updates. These is a big topic too broad for in depth discussion here - but basically you either:
- Drop events. There have been many ways discussed to tackle this in Rx. I current like Ignore incoming stream updates if last callback hasn't finished yet but also see With Rx, how do I ignore all-except-the-latest value when my Subscribe method is running and there are many other discussions of this.
- Signal the producer out of band to tell it to slow down or send conflated updates, or
- You introduce an operator that does in-stream conflation - like a smarter `Buffer` that could compress events to, for example, only include the latest price on a stock item etc. You can author operators that are sensitive to the time that `OnNext` invocations take to process, for example.
See if proper buffering helps first, then think about throttling/conflating events at the source as (a UI can only show so much infomation anway) - then consider smarter conflation as this can get quite complex. https://github.com/AdaptiveConsulting/ReactiveTrader is a good example of a project using some advanced conflation techniques.
Problem
I have tried to simplify my issue by a sample code here. I have a producer thread constantly pumping in data and I am trying to batch it with a time delay between batches so that the UI has time to render it. But the result is not as expected, the produce and consumer seems to be on the same thread. I don't want the batch buffer to sleep on the thread that is producing. Tried `SubscribeOn` did not help much. What am I doing wrong here, how do I get this to print different thread Ids on producer and consumer thread. ``` static void Main(string[] args) { var stream = new ReplaySubject<int>(); Task.Factory.StartNew(() => { int seed = 1; while (true) { Console.WriteLine("Thread {0} Producing {1}", Thread.CurrentThread.ManagedThreadId, seed); stream.OnNext(seed); seed++; Thread.Sleep(TimeSpan.FromMilliseconds(500)); } }); stream.Buffer(5).Do(x => { Console.WriteLine("Thread {0} sleeping to create time gap between batches", Thread.CurrentThread.ManagedThreadId); Thread.Sleep(TimeSpan.FromSeconds(2)); }) .SubscribeOn(NewThreadScheduler.Default).Subscribe(items => { foreach (var item in items) { Console.WriteLine("Thread {0} Consuming {1}", Thread.CurrentThread.ManagedThreadId, item); } }); Console.Read(); } ```