system.reactive - Rx - 几个生产者/一个消费者

Question

一直在尝试用谷歌搜索，但有点卡住了。

假设我们有一个触发事件的类，并且该事件可以由多个线程同时触发。

使用 Observable.FromEventPattern，我们创建一个 Observable，并订阅该事件。 Rx 究竟如何管理同时触发的多个事件？假设我们在不同的线程上快速连续触发了 3 个事件。它是否在内部将它们排队，然后为每个同步调用订阅委托(delegate)？假设我们在一个线程池上订阅，我们还能保证订阅会及时单独处理吗？

接下来，假设对于每个事件，我们想要执行一个 Action ，但它是一个可能不是线程安全的方法，所以我们一次只希望一个线程在这个方法中。现在我看到我们可以使用 EventLoop 调度程序，并且大概我们不需要对代码实现任何锁定？

另外，观察当前线程是否是一种选择？当前线程是触发事件的线程，还是设置订阅的事件？即，当前线程是否保证始终相同，或者可能有 2 个线程同时运行在方法中？

谢谢

PS:我把一个例子放在一起，但我似乎总是在我的订阅方法中结束同一个线程，即使我在线程池上观察，这令人困惑:S

PSS:从做更多的实验来看，如果没有指定调度程序，那么 RX 将只在触发事件的任何线程上执行，这意味着它同时处理几个。一旦我介绍了调度器，它总是连续运行，不管调度器的类型是什么。奇怪:S

Answer 1

根据Rx Design Guidelines , observable 永远不应该调用 OnNext一个观察者同时。在进行下一次调用之前，它将始终等待当前调用完成。全部 Rx方法遵守这个约定。而且，更重要的是，他们假设你也遵守这个约定。当您违反此条件时，您可能会在 Observable 的行为中遇到细微的错误。 .

对于那些不遵守此约定的源数据(即它可以同时生成数据)时，它们提供 Synchronize .
Observable.FromEventPattern假设您不会触发并发事件，因此不会阻止并发下游通知。如果您计划从多个线程触发事件，有时是并发的，那么使用 Synchronize()作为您在 FromEventPattern 之后执行的第一个操作:

// this will get you in trouble if your event source might fire events concurrently.
var events = Observable.FromEventPattern(...).Select(...).GroupBy(...);

// this version will protect you in that case.
var events = Observable.FromEventPattern(...).Synchronize().Select(...).GroupBy(...);

现在，正如 Rx 设计指南所 promise 的那样，所有下游运算符(operator)(最终是您的观察者)都受到了并发通知的保护。 Synchronize通过使用简单的互斥体(又名 lock 语句)来工作。没有花哨的排队或任何东西。如果一个线程试图引发一个事件，而另一个线程已经在引发它，则第二个线程将阻塞，直到第一个线程完成。