将Sample（TimeSpan span）添加到Reactive Extensions管道会导致线程问题

通过使用Reactive Extensions，我创建了一个值的滚动缓冲区，该值在数据流中缓存了一小段最近值的历史记录，供绘图应用程序使用。由于这些值的到达速度比我感兴趣的显示速度快得多，因此我想在我的反应式管道中使用Sample（Timespan span）方法来减慢速度。但是，将其添加到下面的示例中会使WriteEnumerable方法中的某个位（集合被修改）之后引发Exception。显然，这是与Sample有关的线程问题，但是我对如何减轻它感到困惑。我尝试将计划程序设置为在Sample方法中使用无济于事。

有什么建议吗？

class Program
{
    static void Main(string[] args)
    {
        Observable.Interval(TimeSpan.FromSeconds(0.1))
            .Take(500)
            .TimedRollingBuffer(TimeSpan.FromSeconds(10))
            .Sample(TimeSpan.FromSeconds(0.5))
            .Subscribe(frame => WriteEnumerable(frame));

        var input = "";
        while (input != "exit")
        {
            input = Console.ReadLine();
        }
    }

    private static void WriteEnumerable<T>(IEnumerable<T> enumerable)
    {
        foreach (T thing in enumerable)
            Console.WriteLine(thing + " " + DateTime.UtcNow);

        Console.WriteLine(Environment.NewLine);
    }
}

public static class Extensions
{
    public static IObservable<IEnumerable<Timestamped<T>>> TimedRollingBuffer<T>(this IObservable<T> observable, TimeSpan timeRange)
    {
        return Observable.Create<IEnumerable<Timestamped<T>>>(
            o =>
            {
                var queue = new Queue<Timestamped<T>>();
                return observable.Timestamp().Subscribe(
                    tx =>
                    {
                        queue.Enqueue(tx);
                        DateTime now = DateTime.Now;
                        while (queue.Peek().Timestamp < now.Subtract(timeRange))
                            queue.Dequeue();

                        o.OnNext(queue);
                    },
                    ex => o.OnError(ex),
                    () => o.OnCompleted()
                    );
            });
    }
}

贷项到期的地方：被动扩展滑动时间窗口