3

我刚刚开始学习如何使用反应式框架，并且正在努力将多播发布到多个订阅者。

我让一切都像这样正常工作：

m_MessagePublisher = m_ServerClient.MessageQueue
      .GetConsumingEnumerable()
      .ToObservable(TaskPoolScheduler.Default);

var genericServerMessageSubscriber = m_MessagePublisher
      .Where(message => message is GenericServerMessage)
      .Subscribe(message =>
      {
          // do something here
      }

但后来我意识到这不支持多播，当我尝试附加另一个应该被相同消息击中的订阅者时，它不会触发。我一直在阅读 .MultiCast 扩展名并试图弄清楚 Subject 是如何参与其中的，但还不能让它工作：

var subject = new Subject<BesiegedMessage>();

var messagePublisher = m_ServerClient.MessageQueue
      .GetConsumingEnumerable()
      .ToObservable(TaskPoolScheduler.Default)
      .Multicast(subject);

// All generic server messages are handled here
var genericServerMessageSubscriber = subject
      .Where(message => message is GenericServerMessage)
      .Subscribe(message =>
      {
            // do something here
      }

但现在没有一个订阅者受到打击，包括之前工作正常的单个订阅者。为了能够正确多播给多个订阅者，我在这里缺少什么？

更新：使用 Subscribe(subject) 而不是 Multicast(subject) 似乎正在多播，这让我对 .MultiCast() 的用途感到非常困惑

4

1 回答 1

2

编辑：

哈哈-适合我阅读太快-您要问的要简单得多...也就是说，我认为以下内容很重要，所以我要离开了...所以，您的问题-尝试添加此行：

var messagePublisher = m_ServerClient.MessageQueue
  .GetConsumingEnumerable()
  .ToObservable(TaskPoolScheduler.Default)
  .Multicast(subject)
  // Here: connectable observables are a PITA...
  .RefCount();

结束编辑：

嗯...如何描述Multicast...我想我们举个例子：

假设你有这样的东西——你认为它会产生什么？

int delay = 100;
var source = Observable.Interval(TimeSpan.FromMilliseconds(delay));
var publishingFrontend = new Subject<string>();

// Here's "raw"
var rawStream = source;
using(rawStream.Subscribe(x => Console.WriteLine("{0}", x)))
{
    Thread.Sleep(delay * 3);
    using(rawStream.Subscribe(x => Console.WriteLine("Inner: {0}", x)))
    {
        Thread.Sleep(delay * 3);
    }
    Thread.Sleep(delay * 5);
}

由于您订阅的是原始流，因此新订阅者基本上是从头开始的：

（如果您重新运行，这将不会 100% 匹配，因为我采取了笨拙的方式Thread.Sleep，但应该接近）

0
1
2
Inner: 0
3
Inner: 1
4
5
6
7
8
9

嗯……所以如果我们想“在中游”，我们使用以下Publish().RefCount()模式：

var singleSource = source.Publish().RefCount();
using(singleSource.Subscribe(x => Console.WriteLine("{0}", x)))
{
    Thread.Sleep(delay * 3);
    using(singleSource.Subscribe(x => Console.WriteLine("Inner: {0}", x)))
    {
        Thread.Sleep(delay * 3);
    }
    Thread.Sleep(delay * 5);
}

这会产生类似的东西：

0
1
2
Inner: 2
3
Inner: 3
4
Inner: 4
5
6
7
8
9

假设我们没有Publish()操作符——我们如何模拟它？

Console.WriteLine("Simulated Publish:");
// use a subject to proxy values...
var innerSubject = new Subject<long>();
// wire up the source to "write to" the subject
var innerSub = source.Subscribe(innerSubject);
var simulatedSingleSource = Observable.Create<long>(obs =>
{
    // return subscriptions to the "proxied" subject
    var publishPoint = innerSubject.Subscribe(obs);        
    return publishPoint;
});

运行这个，我们得到：

Simulated Publish:
0
1
2
Inner: 2
3
Inner: 3
4
Inner: 4
5
6
7
8
9

哇！

但还有另一种方法...

Console.WriteLine("MulticastPublish:");
var multicastPublish = source.Multicast(new Subject<long>()).RefCount();    
using(multicastPublish.Subscribe(x => Console.WriteLine("{0}", x)))
{
    Thread.Sleep(delay * 3);
    using(multicastPublish.Subscribe(x => Console.WriteLine("Inner: {0}", x)))
    {
        Thread.Sleep(delay * 3);
    }
    Thread.Sleep(delay * 5);
}

输出：

MulticastPublish:
0
1
2
Inner: 2
3
Inner: 3
4
Inner: 4
5
6
7
8
9

编辑：

事实上，所有的ConnectableObservable生成扩展都依赖于Multicast/Subject配对：

Publish() => Multicast(new Subject<T>)
Replay() => Multicast(new ReplaySubject<T>)
PublishLast() => Multicast(new AsyncSubject<T>)
于 2013-04-03T01:35:31.000 回答