2

我是 Rx 的新手，我非常喜欢它。我发现我们代码的现有部分可以通过使用它来大大简化。我遇到了性能问题，如果有一位大师可以帮助我，我将不胜感激。

以前我有一个手动实现的观察者类，它接受订阅和一个关键过滤器。当一个事件进入类时，它将使用提供的键来查找哪些观察者需要回调。这是代码的一个非常简化的版本：

 class OldClass
 {
    private Dictionary<string, List<Action<UsefulInfo>> _callbacks = 
        new Dictionary<string, List<Action<UsefulInfo>>();

    void Subscribe(string key, Action<UsefulInfo> callback)
    {
        _callbacks[key].Add(callback);
    }

    // Some event happens that we want to notify subscribers about
    void EventHandler(object sender, SomeEventArgs e)
    {
        // Create object for callbacks
        UsefulInfo info = CreateUsefulInfo(e);

        string key = info.Key;

        // Look up callbacks for key
        _callbacks[key].ForEach(callback => callback(info));
    }
 }

我已将其更新为使用 Rx，如下所示：

class NewClass
{
    private Subject<UsefulInfo> _subject = new Subject<UsefulInfo>();
    private IObservable<UsefulInfo> _observable;

    public NewClass()
    {
        _observable = _subject.ToObservable();
    }

    IDisposable Subscribe(string key, Action<UsefulInfo> callback)
    {
        return _observable.Where(x => x.Key == key).Subscribe(callback);
    }

    // Some event happens that we want to notify subscribers about
    void EventHandler(object sender, SomeEventArgs e)
    {
        UsefulInfo info = CreateUsefulInfo(e);

        _observable.OnNext(info);
    }
 }

旧代码执行 O(1) 字典键查找以查找回调，但新 Rx 代码调用 Where Func O(n) 次。我有成千上万的观察者。

有没有办法给 Rx 一个 Func，它返回一个键，然后它可以在内部使用它来将观察者存储在字典中？还有其他方法可以提高性能吗？还是我以非预期的方式使用框架？

4

1 回答 1

2

Publish您可以通过使用on序列共享观察者Where'd，然后使用RefCountonIConnectableObservable更智能地管理对源的订阅。

在您的场景中，我想您会将这些“已发布”的 observable 存储在字典中并按需添加。

这是一个基本的实现：

class NewClass 
{ 
    private Subject<UsefulInfo> _subject = new Subject<UsefulInfo>(); 
    private IDictionary<string, IObservable<UsefulInfo>> _keyedObservables; 

    public NewClass() 
    { 
        _keyedObservables = new Dictionary<string, IObservable<UsefulInfo>>();
    } 

    IDisposable Subscribe(string key, Action<UsefulInfo> callback) 
    { 
        // NOT threadsafe for concurrent subscriptions!
        if (!_keyedObservables.Contains(key))
        {
            var keyedAndPublished = _subject.Where(x => x.Key == key)
                .Publish()
                .RefCount();

            _keyedObservables.Add(key, keyedAndPublished);
        }

        return _keyedObservables[key].Subscribe(callback);
    } 

    // Some event happens that we want to notify subscribers about 
    void EventHandler(object sender, SomeEventArgs e) 
    { 
        UsefulInfo info = CreateUsefulInfo(e); 

        _observable.OnNext(info); 
    } 
}
于 2012-06-22T03:03:49.133 回答