RX - 按密钥订阅复杂性的多消费者

问题描述

我得到了 KeyValuePair 的 observable:

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

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

I got a list of consumers defined at runtime.They are only interested in values produced for a single key (myKey).

例如:

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

这是我的消费者订阅代码(每个消费者一个):

Here is my consumer subscription code (one per consumer):

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

让我们来看看:

• N = myObservable
产生的消息数
• M = 消费者数量

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

对于发布的每条新消息，比较将执行 M 次(每个消费者一次).在N条消息的情况下，比较会执行N * M

For every new message being published, Comparison will be executed M times (once per consumer).In the case of N messages, Comparison will be executed N * M

RX 扩展是否提供了另一种避免执行那么多比较的方法?

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

Do I need to make it myself? (using the dictionary of (mykey, consumers) for example, and forwarding the messages to the right consumer(s))

谢谢

推荐答案

你可以试试这个:

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));

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

When I test with this code:

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

对每个键进行一次比较.

The comparison is done once per key.

这篇关于RX - 按密钥订阅复杂性的多消费者的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！