Rx 多个异步操作

Question

我正在阅读有关Rx的信息并浏览一些示例。我正在尝试实现这一点：

List<A> LIstA {get;set;}
List<B> LIstB {get;set;}
List<C> LIstC {get;set;}
private void GetItems()
{
    ListA = GetItemsA();
    ListB = GetItemsB();
    ListC = GetItemsC();
}

所有这些代码都在主线程上执行。我删除了主线程 （UI） 使用此列表的代码。我想要的是异步获取这些项目，但我需要按预定义的顺序获取结果。

以异步方式运行它没有问题，但我在按预定义的顺序获取结果时遇到了问题。因此，我运行代码，显示 UI，几秒钟后填充ListA，然后ListB，最后ListC，仅在此预定义的顺序中。

如何实现这一点？

Answer 1

Task或async可能比Rx更适合此模型。

尽管如此，根据我对你的问题的推断，在前一个完成后使用 concat 连接一个可观察量：

        Func<Action, IObservable<Unit>> fetch = 
            action => Observable.Defer(() => Observable.Start(action));
        fetch(() => A())
        .Concat(fetch(() => B()))
        .Concat(fetch(() => C()))
        .Subscribe();

Answer 2

这对你有用吗？

GetItemsA().ToObservable()
    .Zip(GetItemsB().ToObservable(), (a, b) => new { a, b, })
    .Zip(GetItemsC().ToObservable(), (ab, c) => new { ab.a, ab.b, c, })
    .Subscribe(abc =>
    {
        ListA = abc.a;
        ListB = abc.b;
        ListC = abc.c;
    });

Answer 3

根据您的问题，您有三个不同类型的集合。显然，人们不能将三个不同类型的流连接起来。我认为@Enigmativity是最正确的，但为了使它正常工作，您需要将其更改为：

var uiScheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);
ListA = new ListA();
ListB = new ListB();
ListC = new ListC();
GetItemsA().ToObservable()
.Zip(GetItemsB().ToObservable(), (a, b) => new { a, b, })
.Zip(GetItemsC().ToObservable(), (ab, c) => new { ab.a, ab.b, c, })
.ObserveOn(uiScheduler)
.Subscribe(abc =>
{
    ListA.Add(abc.a);
    ListB.Add(abc.b);
    ListC.Add(abc.c);
});

另一种解决方案可能是：

var a = Observable.Start(() => GetListA());
var b = Observable.Start(() => GetListB());
var c = Observable.Start(() => GetListC());
a.Zip(b, c, (x, y, z) => Tuple.Create(x, y, z))
    .ObserveOn(uiScheduler)
    .Subscribe(result =>
                 {
                   ListA = result.Item1;
                   ListB = result.Item2;
                   ListC = result.Item3;
                 });

我希望集合在创建后立即创建：

  a.ObserveOn(uiScheduler)
    .Do(l => ListA = l)
    .Zip(b, (la, lb) => lb)
    .ObserveOn(uiScheduler)
    .Do(l => ListB = l)
    .Zip(c, (lb, lc) => lc)
    .ObserveOn(uiScheduler)
    .Subscribe(listc => ListC = listc);

Answer 4

你也可以使用动态...：

void Main()
{
    var q =
    from @as in Observable.Start(() => GetItemsA())
    from bs in Observable.Start(() => GetItemsB())
    from cs in Observable.Start(() => GetItemsC())
    select @as.Cast<dynamic>().Concat(bs.Cast<dynamic>()).Concat(cs.Cast<dynamic>());   
    q.Subscribe(t => t.Dump()); 
}
// Define other methods and classes here
private IEnumerable<int> GetItemsA() {
    yield return 1;
    Thread.Sleep(500);
    yield return 2;
    yield return 3;
}
private IEnumerable<string> GetItemsB() {
    yield return "A";
    yield return "B";
    Thread.Sleep(1000);
    yield return "C";
}
private List<float> GetItemsC() {
    return new List<float> { 1.1f, 2.2f, 3.3f };
}