5

我正在寻找一种在不中断的情况下动态合并数据源的方法。现实世界的场景是从多个来源提取类似的数据，而不考虑冗余信息。

为了简化代码，我用一个简单的数字生成器替换了更复杂的代码，它会不断产生数据。这可以与从多个外部服务器读取连续数据流进行比较。

我希望能够合并两个源并将结果（在适当的时候）打印到控制台，这部分效果很好。当我们终止这两个来源并合并到另一个来源时，事情就会按预期停止工作。在这种情况下，我们可以很容易地重新连接合并的StreamObserver，但是，在一个更大的应用程序中，我们必须关注数据的差距以及跟踪订阅了哪些观察者。

有没有解决的办法？

// 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 CancellationTokenSource();
    IObservable<int> xt = Generate(tokenSource2, "B");

    // odd queries
    var seq1 = from n in xs where n % 2 == 1 select n;

    // even queries
    var seq2 = from n in xt where n % 2 == 0 select n;

    // merge everything together
    merged = merged.Merge<int>(seq1);
    merged = merged.Merge<int>(seq2);

    // observer for the merged "streams"
    // NOTE: while this does not appear to be working correctly,
    // remember you have 2 streams and 2 queries at work.  It
    // really is doing what it's expected to here.
    IDisposable mergedStreamObserver = merged.Subscribe(str => { Console.WriteLine(str); });

    // kill both sources
    Console.ReadKey();

    tokenSource1.Cancel();
    tokenSource2.Cancel();

    // start source and query for evens
    // try to merge it
    Console.ReadKey();

    tokenSource2 = new CancellationTokenSource();
    xt = Generate(tokenSource2, "B");

    seq2 = from n in xt where n % 2 == 0 select n;

    merged = merged.Merge(seq2);

    // Nothing is happening because the merged stream was modified.
    // How do we create a composite Observable from multiple sources
    // and dynamically add/terminate those sources?

    Console.ReadKey();

    tokenSource2.Cancel();
    mergedStreamObserver.Dispose();
    Console.ReadKey();
}

static IObservable<int> Generate(CancellationTokenSource tokenSource, string name) {
    Random random = new Random();

    Action<int> observer = _ => { };    /* We could use null, but then at every invocation
                                            * we'd have to copy to a local and check for null.
                                            */

    Task.Factory.StartNew(() => {
            while(!tokenSource.IsCancellationRequested) {
                var t = random.Next(0, 100);
                Console.WriteLine("From Generator {0}: {1}", name, t);

                observer(t);

                Thread.Sleep(1000);
            }

            Console.WriteLine("Random Generator Stopped");
        }, tokenSource.Token);

    return Observable.FromEvent<int>(
        eh => observer += eh,
        eh => observer -= eh);
}
4

1 回答 1

7

使用主题，并在创建流之前订阅合并的流：

var streams = new Subject<IObservable<int>>();
var mergedStreams = streams.Merge();
var mergedObserver = mergedStreams.Subscribe(...);

// now create your streams
...

// add them to the streams subject
streams.OnNext(seq1);
streams.OnNext(seq2);
...

streams.OnNext(seq3);
streams.OnNext(seq4);

...
// If we know there will be no more streams, tell the Subject...
streams.OnCompleted();
于 2013-07-09T17:16:12.407 回答