订阅多个Observable并在同一线程上得到通知

Question

foreach (...) {
    Observable.FromAsync(GetData()/* an async function() */)
        .DoWhile(() => true)
        .Subscribe(data => {
            DoSomething(); // I want all of my Subscribe() to call this on the same thread
        });
}

GetData()将异步返回数据。我创建了多个Observable，对其进行订阅，并希望在返回数据时在同一线程上得到通知（调用DoSomething()）。我已经尝试过BlockingCollection<>，但徒劳无功...：（

有人知道该怎么做吗？

PS：我不想使用消息队列（例如Redis pub / sub）来实现这一目标。

Answer 1

您要处理同一线程上的每条消息，并阻塞线程，直到下一条消息到达为止-这是一种非常常见的范例，称为Event Loop。

事实证明，您可以在Rx的事件循环中使用适当的名称EventLoopScheduler运行事物。

    var values = Observable.Interval(TimeSpan.FromSeconds(0.1));

    void DoSomething(long value) => Console.WriteLine($"Value: {value}, Thread: {Thread.CurrentThread.Name}");

    var eventloop = new EventLoopScheduler();

    values.ObserveOn(eventloop).Subscribe(DoSomething);

输出：

Value: 0, Thread: Event Loop 1
Value: 1, Thread: Event Loop 1
Value: 2, Thread: Event Loop 1
Value: 3, Thread: Event Loop 1
Value: 4, Thread: Event Loop 1
Value: 5, Thread: Event Loop 1
Value: 6, Thread: Event Loop 1
Value: 7, Thread: Event Loop 1