响应式扩展(Rx) GroupByUntil死锁

Question

我试图节流CarUpdatedEventHandler来自CarModel并将它们发送回我的MapListModel。我想出了一些方法，但不幸的是，由于未知的原因，这导致我死锁。

这段代码的行为是正确的，但导致死锁，所以我的问题是我还能做什么?

var observable = Observable.FromEventPattern<CarUpdatedEventAgrs>(_carModel, "CarUpdatedEventHandler").GroupByUntil(
                key => key.EventArgs.Car, grp => grp.Throttle(TimeSpan.FromMilliseconds(WATCHLIST_THROTTLE_DEPLAY_MS)));
observable.Subscribe(i =>
{
   CarUpdatedEventAgrs e = new CarUpdatedEventAgrs();
   e.Car = i.Key;
   this.RaiseEvent<CarUpdatedEventAgrs>(MapListUpdateEventHandler, e);
});

感谢您的宝贵时间!

此代码也导致死锁:

var observable = Observable.FromEventPattern<CarUpdatedEventAgrs>(_carModel, "CarUpdatedEventHandler").GroupByUntil(
                    key => key.EventArgs.Car, grp => grp.Throttle(TimeSpan.FromMilliseconds(WATCHLIST_THROTTLE_DEPLAY_MS)));
    observable.Subscribe(i =>
    {
       Console.WriteLine("Wait for deadlock");
    });

Answer 1

我认为这个问题与节气门操作符作用于与调度程序使用的主线程不同的线程有关，因此您可能会遇到一些问题，特别是如果您试图以某种方式更新UI。您应该考虑在订阅之前使用ObserveOn操作符，并可能在主UI线程上进行观察。假设您正在开发一个WPF应用程序，可以这样做:

observable.ObserveOnDispatcher().Subscribe( // rest of the code

为了做到这一点，你需要Rx-WPF包。显然，如果你使用的是Windows窗体，你将需要相应的调度程序。

Answer 2

最后能够分组和节流，下面是使用Rx的方法:

var throttledEvent = Observable.FromEventPattern<YourEventArgs>(ev => _yourObject.SubscribeEvent += ev, ev => _yourObject.SubscribeEvent -= ev).GroupBy(key => key.EventArgs.YourProperty).Select(o => o.Throttle(TimeSpan.FromMilliseconds(YourThrottleTimeMs))).Merge();

这样很好!