6

我有两个对象流，每个对象都有一个Timestamp值。两个流都是有序的，因此例如时间戳可能是一个流中的 T a =1,3,6,6,7和另一个流中的 T b = 1,2,5,5,6,8。两个流中的对象属于同一类型。

我想做的是将这些事件中的每一个按时间戳顺序放置在总线上，即放置 A 1，然后放置 B 1、 B 2、 A 3等等。此外，由于某些流具有多个具有相同时间戳的（顺序）元素，因此我希望将这些元素分组，以便每个新事件都是一个数组。所以我们将 [A 3 ] 放在总线上，然后是 [A 1 5 ,A 2 5 ] 等等。

我试图通过创建两个ConcurrentQueue结构来实现这一点，将每个事件放在队列的后面，然后查看队列的每个前面，首先选择较早的事件，然后遍历队列，以便存在具有此时间戳的所有事件.

但是，我遇到了两个问题：

  • 如果我让这些队列不受限制，我很快就会耗尽内存，因为读取操作比接收事件的处理程序快得多。（我有几千兆字节的数据）。
  • 我有时会遇到这样一种情况，即我在 A 2 5到达之前处理事件，例如 A 1 5 。我不知何故需要提防这一点。

我认为 Rx 可以在这方面提供帮助，但我没有看到明显的组合器可以使这成为可能。因此，非常感谢任何建议。

4

1 回答 1

10

Rx 确实非常适合这个问题 IMO。

IObservables由于显而易见的原因（您必须首先观察整个流以保证正确的输出顺序），不能“OrderBy”，所以我在下面的回答假设（您所说的）您的 2 个源事件流是有序的。

最后这是一个有趣的问题。标准的 Rx 运算符缺少一个GroupByUntilChanged可以轻松解决这个问题的方法，只要它OnComplete在观察到下一组的第一个元素时调用前一组 observable 即可。但是，查看DistinctUntilChanged它的实现并不遵循这种模式，并且仅OnComplete在源 observable 完成时调用（即使它知道在第一个非不同元素之后将不再有元素......奇怪？？？）。无论如何，出于这些原因，我决定反对一种GroupByUntilChanged方法（不违反 Rx 约定），而是选择了ToEnumerableUntilChanged.

免责声明：这是我的第一个 Rx 扩展，因此希望对我的选择提供反馈。此外，我主要关心的一个问题是持有该distinctElements列表的匿名 observable。

首先，您的应用程序代码非常简单：

    public class Event
    {
        public DateTime Timestamp { get; set; }
    }

    private IObservable<Event> eventStream1;
    private IObservable<Event> eventStream2; 

    public IObservable<IEnumerable<Event>> CombineAndGroup()
    {
        return eventStream1.CombineLatest(eventStream2, (e1, e2) => e1.Timestamp < e2.Timestamp ? e1 : e2)
            .ToEnumerableUntilChanged(e => e.Timestamp);
    }

现在ToEnumerableUntilChanged执行（代码警告墙）：

    public static IObservable<IEnumerable<TSource>> ToEnumerableUntilChanged<TSource,TKey>(this IObservable<TSource> source, Func<TSource,TKey> keySelector)
    {
        // TODO: Follow Rx conventions and create a superset overload that takes the IComparer as a parameter
        var comparer = EqualityComparer<TKey>.Default;

        return Observable.Create<IEnumerable<TSource>>(observer =>
        {
            var currentKey = default(TKey);
            var hasCurrentKey = false;
            var distinctElements = new List<TSource>();

            return source.Subscribe((value =>
            {
                TKey elementKey;
                try
                {
                    elementKey = keySelector(value);
                }
                catch (Exception ex)
                {
                    observer.OnError(ex);
                    return;
                }

                if (!hasCurrentKey)
                {
                    hasCurrentKey = true;
                    currentKey = elementKey;
                    distinctElements.Add(value);
                    return;
                }

                bool keysMatch;
                try
                {
                    keysMatch = comparer.Equals(currentKey, elementKey);
                }
                catch (Exception ex)
                {
                    observer.OnError(ex);
                    return;
                }

                if (keysMatch)
                {
                    distinctElements.Add(value);
                    return;
                }

                observer.OnNext( distinctElements);

                distinctElements.Clear();
                distinctElements.Add(value);
                currentKey = elementKey;

            }), observer.OnError, () =>
            {
                if (distinctElements.Count > 0)
                    observer.OnNext(distinctElements);

                observer.OnCompleted();
            });
        });
    }
于 2012-03-17T01:31:40.783 回答