RX -按关键订阅复杂性划分的多个使用者

Question

我得了observable of KeyValuePair<int,double>

--(7|45.2)--(3|11.1)--(5|13.2)--(6|36.2)--(3|57.4)

我得到了一个在运行时定义的使用者列表。他们只对为单个键(myKey)生成的值感兴趣。

例如：

• 消费者7，只对价值45.2感兴趣。
• 消费者3，只对11.1和57.4的数值感兴趣。
• 消费者1只对myKey = 1的值感兴趣，所以这里没有。

以下是我的用户订阅代码(每个用户一个)：

myObservable.Where(t => t.Key == myKey).Subscribe(t => /* DoSomething*/);

我们来看看：

• N= myObservable生成的消息数
• M=消费者人数

让我们调用比较代码t.Key == myKey

对于发布的每一条新消息，将执行M次比较(每个使用者一次)。对于N个消息，比较将执行N*M

是否提供了另一种方法来避免执行如此多的比较？

我需要自己做吗？(例如，使用(mykey，consumers)字典，并将消息转发给正确的使用者)

谢谢

Answer 1

因此，您希望优化此设置：

IObservable<KeyValuePair<int, double>> myObservable;

myObservable.Where(e => e.Key == 1).Select(e => e.Value).Subscribe(x => { /* DoSomething*/ });
myObservable.Where(e => e.Key == 3).Select(e => e.Value).Subscribe(x => { /* DoSomething*/ });
myObservable.Where(e => e.Key == 7).Select(e => e.Value).Subscribe(x => { /* DoSomething*/ });

此设置确实会导致对myObservable发出的每个元素进行多个比较。优化它的一个简单方法是将myObservable与GroupBy操作符分组，以便创建一个嵌套的可观察结构：

IObservable<IGroupedObservable<int, double>> myGroupedObservable = myObservable
    .GroupBy(e => e.Key, e => e.Value);

然后，...and选择要观察的组，并将嵌套结构Merge为平面序列：

myGroupedObservable.Where(g => g.Key == 1).Merge().Subscribe(x => { /* DoSomething*/ });
myGroupedObservable.Where(g => g.Key == 3).Merge().Subscribe(x => { /* DoSomething*/ });
myGroupedObservable.Where(g => g.Key == 7).Merge().Subscribe(x => { /* DoSomething*/ });

这样，每键和使用者只能进行一次比较。现在您正在比较IGroupedObservable<int, double>子序列，而不是KeyValuePair<int, double>元素。只有当GroupBy操作符接收到具有不同键的第一个元素时，才会发出这些子序列。因此，比较的总数将等于NxM，其中N是使用者的数目，M是不同的键数。除非您与数以百万计的消费者打交道，否则这个数字应该相当小，不应造成明显的开销。

基于GroupBy运算符和ConcurrentDictionary<TKey, Subject<TSource>>组合的更复杂的方法可以在这个答案的5th revision中找到。它可能比上面的简单方法效率低，因为其中涉及两个字典( GroupBy也在内部使用一个字典)，所以每个键被散列两次。

Answer 2

你可以试试：

var subject = new Subject<KeyValuePair<int, double>>();

var interests = new Dictionary<int, Func<double, string>>()
{
    { 7, v => $"Seven:{v}" },
    { 3, v => $"Three:{v}" },
    { 1, v => $"One:{v}" },
};

IDisposable subscription =
    subject
        .GroupBy(x => x.Key, x => x.Value)
        .Select(gx =>
            interests.ContainsKey(gx.Key)
            ? gx.Select(x => interests[gx.Key](x))
            : Observable.Empty<string>())
        .Merge()
        .Subscribe(x => Console.WriteLine(x));

当我使用此代码进行测试时：

subject.OnNext(new KeyValuePair<int, double>(7, 45.2));
subject.OnNext(new KeyValuePair<int, double>(3, 11.1));
subject.OnNext(new KeyValuePair<int, double>(5, 13.2));
subject.OnNext(new KeyValuePair<int, double>(6, 36.2));
subject.OnNext(new KeyValuePair<int, double>(3, 57.4));

我得到了这个输出：

Seven:45.2
Three:11.1
Three:57.4

每键进行一次比较。