我正在编写一个用于聚合时间事件以进行报告的C# (.NET 4.5)应用程序。为了使我的查询逻辑可重用于实时和历史数据,我利用了Reactive Extensions (2.0)及其IScheduler基础设施(HistoricalScheduler等)。
例如,假设我们创建了一个事件列表(按时间顺序排序,但它们可能重叠!),其唯一负载是它们的时间戳,并想要知道它们在固定持续时间的缓冲区中的分布:
const int num = 100000;
const int dist = 10;
var events = new List<DateTimeOffset>();
var curr = DateTimeOffset.Now;
var gap = new Random();
var time = new HistoricalScheduler(curr);
for (int i = 0; i < num; i++)
{
events.Add(curr);
curr += TimeSpan.FromMilliseconds(gap.Next(dist));
}
var stream = Observable.Generate<int, DateTimeOffset>(
0,
s => s < events.Count,
s => s + 1,
s => events[s],
s => events[s],
time);
stream.Buffer(TimeSpan.FromMilliseconds(num), time)
.Subscribe(l => Console.WriteLine(time.Now + ": " + l.Count));
time.AdvanceBy(TimeSpan.FromMilliseconds(num * dist));
运行此代码会导致一个System.StackOverflowException
,并显示以下堆栈跟踪(最后3行):
运行此代码会导致System.StackOverflowException
,并显示以下堆栈跟踪(最后3行):
mscorlib.dll!System.Threading.Interlocked.Exchange<System.IDisposable>(ref System.IDisposable location1, System.IDisposable value) + 0x3d bytes
System.Reactive.Core.dll!System.Reactive.Disposables.SingleAssignmentDisposable.Dispose() + 0x37 bytes
System.Reactive.Core.dll!System.Reactive.Concurrency.ScheduledItem<System.DateTimeOffset>.Cancel() + 0x23 bytes
...
System.Reactive.Core.dll!System.Reactive.Disposables.AnonymousDisposable.Dispose() + 0x4d bytes
System.Reactive.Core.dll!System.Reactive.Disposables.SingleAssignmentDisposable.Dispose() + 0x4f bytes
System.Reactive.Core.dll!System.Reactive.Concurrency.ScheduledItem<System.DateTimeOffset>.Cancel() + 0x23 bytes
...
好的,问题似乎出在我使用了Observable.Generate()
,这取决于列表大小(num
),无论选择哪种调度程序。
我做错了什么?或者更一般地说,从提供自己时间戳的事件的IEnumerable
创建IObservable
的首选方法是什么?
num
可以有多大?此外,如果您在调试器中逐步执行此操作,那么在看到错误之前执行的最后一行代码是什么? - Jim Mischelnum = 51600
(在 Release 配置中稍微少一些,在 Debug 配置中稍微多一些)。可观察序列似乎已经完全创建。我可以在Observable.Generate()
的 lambda 表达式处设置断点。异常在最后一次调用Console.WriteLine()
后抛出。 - bastian schmickCancel
或Dispose
的递归调用,这将使您的堆栈溢出(默认大小为1兆字节)。我不太熟悉Observable
,无法说明为什么会发生这种情况。 - Jim Mischel