gpt4 book ai didi

c# - 合并两个 Observable 时保留排序

转载 作者:行者123 更新时间:2023-11-30 17:26:04 26 4
gpt4 key购买 nike

我想合并 2 个可观察对象并保持顺序(可能基于选择器)。我还想对 observable 的来源施加反压。

因此选择器会选择其中一个项目通过可观察对象继续推进,而另一个项目将等待另一个项目也来进行比较。

Src1、Src2 和 Result 都是 IObservable<T> 类型.

Src1: { 1,3,6,8,9,10 }
Src2: { 2,4,5,7,11,12 }
Result: 1,2,3,4,5,6,7,8,9,10,11,12

Timeline:
Src1: -1---3----6------8----9-10
Src2: --2-----4---5-7----11---------12
Result: --1--2--3-4-5-6--7-8--9-10-11-12
  1. 在上面的示例中,src1 发出“1”并被阻塞,直到 src2 发出这是第一项,“2”。
  2. 应用一个选择器,它选择最小的项目从 src1 中选择项目。
  3. Src2 现在等待下一个项目(来自 src1)与其进行比较当前项目('2')。
  4. 当 src1 发出下一个项目“3”时,选择再次运行,这从 src2 中选择项目的时间。
  5. 重复此过程,直到其中一个可观察对象完成。然后,剩余的可观察对象会推送项目,直到它完成。

这是否可以通过现有的 .net Rx 方法实现?

编辑:注意 2 个源可观察量保证是有序的。

示例测试:

var source1 = new List<int>() { 1, 4, 6, 7, 8, 10, 14 }.AsEnumerable();
var source2 = new List<int>() { 2, 3, 5, 9, 11, 12, 13, 15 }.AsEnumerable();

var src1 = source1.ToObservable();
var src2 = source2.ToObservable();

var res = src1.SortedMerge(src2, (a, b) =>
{
if (a <= b)
return a;
else
return b;
});

res.Subscribe((x) => Console.Write($"{x}, "));

期望结果:1,2,3,4,5,6,7,8,9,10,11,12,13,14,15

最佳答案

这很有趣。不得不稍微调整一下算法。它可以进一步改进。

假设:

  1. 有两个流,streamA , streamB普通型T .
  2. 这两个流分别排序为 streamA[i] < streamA[i+1]streamB[i] < stream[i+1] .
  3. 您不能假定 streamA[i] 之间有任何关系和 streamB[i] .
  4. 流 A 和 B 是谨慎的:不会从两者发出相同的元素。如果发生这种情况,我将抛出 NotImplementedException . 这种情况很容易处理,但我想避免歧义。
  5. 有一个函数min对于类型 T .
  6. 没有对两个流的相对速度做出任何假设,但如果一个流始终比另一个快,背压就会成为一个问题。

这是我使用的算法:

  • 有两个队列,qAqB .
  • 当您从 streamA 获得商品时, 将其排入 qA .
  • 当您从 streamB 获得商品时, 将其排入 qB .
  • 虽然两者中都有一个项目qAqB ,比较两个队列的顶部项目。移除并发出这两者的最小值。如果两个队列仍然非空,则重复。
  • 如果streamAstreamB完成,转储队列的内容并终止。 注意:这确实是懒惰的,应该更改为转储,然后继续返回未完成的可观察对象

代码如下:

public static IObservable<T> SortedMerge<T>(this IObservable<T> source, IObservable<T> other)
{
return SortedMerge(source, other, (a, b) => Enumerable.Min(new[] { a, b}));
}

public static IObservable<T> SortedMerge<T>(this IObservable<T> source, IObservable<T> other, Func<T, T, T> min)
{
return source
.Select(i => (key: 1, value: i)).Materialize()
.Merge(other.Select(i => (key: 2, value: i)).Materialize())
.Scan((qA: ImmutableQueue<T>.Empty, qB: ImmutableQueue<T>.Empty, exception: (Exception)null, outputMessages: new List<T>()),
(state, message) =>
{
if (message.Kind == NotificationKind.OnNext)
{
var key = message.Value.key;
var value = message.Value.value;
var qA = state.qA;
var qB = state.qB;
if (key == 1)
qA = qA.Enqueue(value);
else
qB = qB.Enqueue(value);
var output = new List<T>();
while(!qA.IsEmpty && !qB.IsEmpty)
{
var aVal = qA.Peek();
var bVal = qB.Peek();
var minVal = min(aVal, bVal);
if(aVal.Equals(minVal) && bVal.Equals(minVal))
throw new NotImplementedException();

if(aVal.Equals(minVal))
{
output.Add(aVal);
qA = qA.Dequeue();
}
else
{
output.Add(bVal);
qB = qB.Dequeue();
}
}
return (qA, qB, null, output);
}
else if (message.Kind == NotificationKind.OnError)
{
return (state.qA, state.qB, message.Exception, new List<T>());
}
else //message.Kind == NotificationKind.OnCompleted
{
var output = state.qA.Concat(state.qB).ToList();
return (ImmutableQueue<T>.Empty, ImmutableQueue<T>.Empty, null, output);
}
})
.Publish(tuples => Observable.Merge(
tuples
.Where(t => t.outputMessages.Any() && (!t.qA.IsEmpty || !t.qB.IsEmpty))
.SelectMany(t => t.outputMessages
.Select(v => Notification.CreateOnNext<T>(v))
.ToObservable()
),
tuples
.Where(t => t.outputMessages.Any() && t.qA.IsEmpty && t.qB.IsEmpty)
.SelectMany(t => t.outputMessages
.Select(v => Notification.CreateOnNext<T>(v))
.ToObservable()
.Concat(Observable.Return(Notification.CreateOnCompleted<T>()))
),
tuples
.Where(t => t.exception != null)
.Select(t => Notification.CreateOnError<T>(t.exception))
))
.Dematerialize();

ImmutableQueue来自System.Collections.Immutable . Scan需要跟踪状态。因为 OnCompleted 需要实现处理。诚然,这是一个复杂的解决方案,但我不确定是否有更清晰的以 Rx 为中心的方式。

如果您需要进一步说明,请告诉我。

关于c# - 合并两个 Observable 时保留排序,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/57238239/

26 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com