小编cod*_*ape的帖子

我正在寻找一种不间断动态合并数据源的方法.现实世界的情况是在不考虑冗余信息的情况下从多个来源获取数据.

为了简化代码,我用一个简单的数字生成器替换了更复杂的代码,它将持续生成数据.这可以比较来自多个外部服务器的连续数据流.

我希望能够合并两个源并将结果(在适当的时候)打印到控制台,这部分非常有用.当我们终止这两个源并在另一个源中合并时,事情就会按预期停止工作.在这种情况下,我们可以轻松地重新连接mergedStreamObserver,但是,在更大的应用程序中,我们必须关注数据中的差距以及跟踪Observers订阅的内容.

有没有解决的办法？

// imports
using System;
using System.Reactive.Linq;
using System.Threading;
using System.Threading.Tasks;

static void Main(string[] args) {
    // base "stream of results" as we will want to randomly add (and terminate other sources)
    IObservable<int> merged = Observable.Empty<int>();

    // source 1
    var tokenSource1 = new CancellationTokenSource();
    IObservable<int> xs = Generate(tokenSource1, "A");

    // to avoid generating the same numbers, which does happen,
    // sleep some amount of time before calling generate again
    Thread.Sleep(100);

    // source 2
    var tokenSource2 = new …