如何使用Rx合并到数组

Question

我需要使用UniRx合并2个数组，以便得到可观察的，它发出数组的第一个元素，然后是第二个元素，依此类推，然后发出最长数组的其余部分

我尝试了Zip，但是Zip减少了最长数组的尾部，我尝试用Scheduler.DefaultSchedulers.Iteration合并，但是它启动了一个我不想要的并行线程

var x1 = new[] {1, 2, 3}.ToObservable();
var x2 = new[] {4, 5, 6, 7, 8, 9}.ToObservable();
var merge = x1.Merge(x2);
merge.Subscribe(i => print(i));

我以为是1 4 2 5 3 6 7 8 9我得到了1 2 3 4 5 6 7 8 9

Answer 1

正如您所期望的那样，这对我很有效：

var a1 = new int[] { 1, 2, 3 };
var a2 = new int[] { 4, 5, 6, 7, 8, 9 };

var x1 = a1.Select(x => (int?)x).ToObservable().Concat(Observable.Repeat((int?)null));
var x2 = a2.Select(x => (int?)x).ToObservable().Concat(Observable.Repeat((int?)null));

var query =
    x1
        .Zip(x2, (i1, i2) => new [] { i1, i2 })
        .TakeWhile(xs => !(xs[0] == null && xs[1] == null))
        .SelectMany(xs => xs)
        .Where(x => x != null)
        .Select(x => x.Value);

​

​

Answer 2

如果您知道数组的长度，可以将Zip序列与其他两个序列连接起来，并跳过已压缩的元素的数量。这是C#，但应该会让您有所了解：

var a1 = new int[] { 1, 2, 3 };
var a2 = new int[] { 4, 5, 6, 7, 8, 9 };

var x1 = a1.ToObservable();
var x2 = a2.ToObservable();

int skip = Math.Min(a1.Length, a2.Length);
Observable.Zip(x1, x2).SelectMany(x => x)
    .Concat(x1.Skip(skip))
    .Concat(x2.Skip(skip))
    .Subscribe(i => Console.WriteLine(i));

谢谢，此代码适用于数组。但是，在无法访问Length的情况下，如何为通用IObservable<int>实现此操作？

您可以随时计算压缩后的可观察对象中的元素数量：

var x1 = new int[] { 1, 2, 3 }.ToObservable();
var x2 = new int[] { 4, 5, 6, 7, 8, 9 }.ToObservable();

var zip = Observable.Zip(x1, x2);
int skip = await zip.Count().LastAsync();

zip.SelectMany(x => x)
    .Concat(x1.Skip(skip))
    .Concat(x2.Skip(skip))
    .Subscribe(i => Console.WriteLine(i));

Answer 3

@mm8发布的解决方案将使用两个具有定义结束的可观察对象(就像伪装成可观察对象的数组)，但将永远挂起无限个可观察对象，这是一种常见的情况。例如：

var odds = Observable.Interval(TimeSpan.FromMilliseconds(10))
    .Select(i => i * 2 + 1);
var evens = Observable.Interval(TimeSpan.FromMilliseconds(10))
    .Select(i => i * 2);

var zip = Observable.Zip(evens, odds);
int skip = await zip.Count().LastAsync();

// Hangs forever here.

如果你想交错数组，最好的方法就是交错为数组。Rx是用来处理可观察对象的。这不是一个可观察到的问题。