c# - 极长的 Rx 事件链

Question

我正在研究可以最好地描述为模拟/工作流/游戏逻辑引擎的东西(尽管它不完全属于这些类别中的任何一个)。

它的意图是完全由事件驱动(响应式(Reactive))，并且必须支持数万甚至数十万个链式事件的可能性，具有分支、过滤和并发性，以及所有 Rx 优点。

我对 Reactive Extensions 非常不熟悉，因此决定编写我能想到的最简单的测试(将一堆 ISubject 链接在一起)。我很快发现将太多事件(在我的例子中大约有 12000 个)链接在一起会导致 StackOverflowException —— 这对我来说很有意义，因为考虑到 Rx 实际上只是以新颖的方式将事件处理程序连接在一起，调用堆栈只能变得如此深。

所以我正在寻找一种(Reactive-ish？)绕过这个限制的方法。我不可能是唯一一个想用这个框架做一些非常大的事情的人。社区可以提供的任何帮助将不胜感激。

这是我的测试代码:

class Program
{
    static void Main(string[] args)
    {
        for (int i = 0; i < 1000000; i += 1000)
        {
            Console.Write($"{i} ");
            using (Dynamite dynamite = new Dynamite())
            {
                dynamite.Setup(i);
                dynamite.Trigger();
            }
        }
        Console.ReadKey();

    }
}

public class Dynamite : IDisposable
{
    ISubject<bool> start = null;
    IList<IDisposable> handles = new List<IDisposable>();

    public void Setup(int length)
    {
        length = length == 0 ? 1 : length;

        var fuses =
            Enumerable.Range(0, length)
            .Select(v => new Subject<bool>())
            .ToArray();

        ISubject<bool> prev = null;

        foreach (var fuse in fuses)
        {
            //Console.Write(".");

            if (prev != null)
            {
                Attach(prev, fuse);
            }
            prev = fuse;
        }

        start = fuses.First();
        var end = fuses.Last();

        handles.Add(
            end
                .Subscribe(onNext: b =>
                {
                    //Console.Write("t");
                    this.Explode();
                }));
    }

    void Attach(ISubject<bool> source, ISubject<bool> dest)
    {
        var handle = source
            .Subscribe(onNext: b =>
             {
                 //Console.Write("s");
                 dest.OnNext(b);
             });
        handles.Add(handle);
    }

    public void Trigger()
    {
        //Console.Write("p");
        start.OnNext(true);
    }

    void Explode()
    {
        Console.WriteLine("...BOOM!");
    }

    public void Dispose()
    {
        foreach (var h in handles)
            h.Dispose();
    }
}

这是控制台输出:

0 ...BOOM!
1000 ...BOOM!
2000 ...BOOM!
3000 ...BOOM!
4000 ...BOOM!
5000 ...BOOM!
6000 ...BOOM!
7000 ...BOOM!
8000 ...BOOM!
9000 ...BOOM!
10000 ...BOOM!
11000 ...BOOM!
12000
Process is terminated due to StackOverflowException.

Answer 1

我找到了解决方案。 CurrentThread调度程序。

.ObserveOn(Scheduler.CurrentThread)

The CurrentThreadScheduler [...] will schedule actions to be performed on the thread that makes the original call. The action is not executed immediately, but is placed in a queue and only executed after the current action is complete.