我正在研究一些最好被描述为模拟/工作流/游戏逻辑引擎的东西(尽管它并不属于任何这些类别).

其目的是完全由事件驱动(被动),并且必须支持数十甚至数十万个链式事件的可能性,具有分支,过滤和并发性以及所有Rx优点.

我对Reactive Extensions很新,并决定编写我能想到的最简单的测试(将一堆ISubject链接在一起).我很快发现将太多事件链接在一起(在我的情况下约为12000)会导致*Exception – 这在我考虑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 *Exception.

解决方法:

我找到了解决方案. 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.