c# – Reactive Extensions SelectMany包含大对象

发布时间：2020-12-15 23:34:15 所属栏目：百科来源：网络整理

导读：我有一小段代码模拟使用大对象的流(即巨大的byte []).对于序列中的每个项目,调用异步方法以获得一些结果.问题？实际上,它会抛出OutOfMemoryException. 代码与LINQPad(C#程序)兼容： void Main(){ var selectMany = Enumerable.Range(1,100) .Select(i = new

我有一小段代码模拟使用大对象的流(即巨大的byte []).对于序列中的每个项目,调用异步方法以获得一些结果.问题？实际上,它会抛出OutOfMemoryException.

代码与LINQPad(C#程序)兼容：

void Main()
{
    var selectMany = Enumerable.Range(1,100)
                   .Select(i => new LargeObject(i))
                   .ToObservable()
                   .SelectMany(o => Observable.FromAsync(() => DoSomethingAsync(o)));

    selectMany
        .Subscribe(r => Console.WriteLine(r));
}


private static async Task<int> DoSomethingAsync(LargeObject lo)
{
    await Task.Delay(10000);
    return lo.Id;
}

internal class LargeObject
{
    public int Id { get; }

    public LargeObject(int id)
    {
        this.Id = id;
    }

    public byte[] Data { get; } = new byte[10000000];
}

它似乎同时创建了所有对象.我怎么能以正确的方式做到这一点？

基本思想是调用DoSomethingAsync以获得每个对象的一些结果,这就是我使用SelectMany的原因.为了简化,我刚刚介绍了一个Task.Delay,但在现实生活中它是一个可以同时处理一些项的服务,所以我想引入一些并发机制来获得它的优势.

请注意,从理论上讲,处理少量项目不应该填满内存.实际上,我们只需要每个“大对象”来获取DoSomethingAsync方法的结果.在那之后,不再使用大对象.

解决方法

我觉得我是 repeating myself.与你的上一个问题和我的上一个答案类似,你需要做的是限制同时创建的bigObjects?的数量.

为此,您需要将对象创建和处理组合在一起并将其放在同一个线程池中.现在问题是,我们使用异步方法允许线程在我们的异步方法运行时执行其他操作.由于您的慢速网络调用是异步的,因此您的(快速)对象创建代码将继续过快地创建大型对象.

相反,我们可以使用Rx通过将对象创建与异步调用相结合来保持运行的并发Observable的数量,并使用.Merge(maxConcurrent)来限制并发.

作为奖励,我们还可以设置查询执行的最短时间.就在Zip,只需要很短的延迟.

static void Main()
{
    var selectMany = Enumerable.Range(1,100)
                        .ToObservable()
                        .Select(i => Observable.Defer(() => Observable.Return(new LargeObject(i)))
                            .SelectMany(o => Observable.FromAsync(() => DoSomethingAsync(o)))
                            .Zip(Observable.Timer(TimeSpan.FromMilliseconds(400)),(el,_) => el)
                        ).Merge(4);

    selectMany
        .Subscribe(r => Console.WriteLine(r));

    Console.ReadLine();
}


private static async Task<int> DoSomethingAsync(LargeObject lo)
{
    await Task.Delay(10000);
    return lo.Id;
}

internal class LargeObject
{
    public int Id { get; }

    public LargeObject(int id)
    {
        this.Id = id;
        Console.WriteLine(id + "!");
    }

    public byte[] Data { get; } = new byte[10000000];
}

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!