SelectMany 使用 ReactiveExtensions 占用大量内存答案

【问题标题】：SelectMany taking a huge amount of memory using ReactiveExtensionsSelectMany 使用 ReactiveExtensions 占用大量内存
【发布时间】：2016-05-08 13:28:41
【问题描述】：

我想创建一个获取图像并返回一些派生对象的管道。

我正在使用一系列位图，并为每个位图执行任务（即异步）。所以它看起来很简单。但是，我发现内存消耗真的很高。为了说明问题，我创建了这个可以运行的测试。

请看一下内存，因为它会占用多达 400 MB 的 RAM。

我可以做些什么来避免占用这么多内存？这里发生了什么？

[Fact]
public async Task BitmapPipelineTest()
{
    var bitmaps = Enumerable.Range(0, 100).Select(_ => new WriteableBitmap(800, 600, 96, 96, PixelFormats.Bgr24, new BitmapPalette(new List<Color>() { new Color() })));
    var bitmapsObs = bitmaps.ToObservable();

    var processed = bitmapsObs.SelectMany(bitmap => DoSomethingAsync(bitmap));
    processed.Subscribe();

    await Task.Delay(20000);
}

private async Task<object> DoSomethingAsync(BitmapSource bitmap)
{
    await Task.Delay(1000);
    return new object();
}

【问题讨论】：

如果你搜索“WriteableBitmap”和“内存泄漏”，有很多发帖者遇到同样的问题，并且缺乏好的解决方案。如果您不需要使用 WriteableBitmap（即，如果您不做 UI 工作），那么我将使用 Bitmap 类，并修改您的代码以使用 SelectMany 中的 using 语句创建/处理 Bitmap跨度>
@Andrew - 这可能与您描述的问题有关。我认为这取决于 OP 使用的 .NET 版本。看起来内存问题在 4.0 中得到了解决。在我使用 .NET 4.5 进行的测试中，内存确实会迅速失控，但它最终会被垃圾回收，所以我认为没有内存泄漏。
@JasonBoyd - 有趣的是，使用 4.5，我能够破坏最大内存分配 (2Gb) 并使进程崩溃。在位图超出我修改的代码的范围后调用 GC.Collect() 也没有阻止它（尽管我永远不会考虑将这样的 GC 调用投入生产）
@Andrew - 是的，由于 cmets 中的空间有限，当我说“它最终确实会收集垃圾”时，我掩饰了很多。事实上，如果我让它不加检查地运行，我的应用程序也会崩溃。但是，我可以通过直接调用 GC.Collect(); GC.WaitForPendingFinalizers(); GC.Collect(); 来释放内存，这向我表明这不是内存泄漏，而是 GC 计划的问题（这是我在回答中提到的）。我还能够通过在按钮单击事件处理程序后面以 200 个批次运行 GC 来触发它们。

标签： c# .net system.reactive reactive-programming

【解决方案1】：

所以我认为问题不一定是由于SelectMany 甚至是响应式扩展。看起来WriteableBitmap 使用非托管内存：source code。我认为问题在于，您正在以非常快的速度连续创建一堆相对较小的托管对象，这些对象占用了大量的非托管内存。来自MSDN：

如果一个小的托管对象分配了大量的非托管内存，运行时只考虑托管内存，从而低估了调度垃圾回收的紧迫性。

但是我们可以使用GC.AddMemoryPressure 和GC.RemoveMemoryPressure 函数给垃圾收集器提示。这将有助于 GC 改进其调度。在我们这样做之前，我们需要对分配的非托管内存量有一些了解。我相信非托管内存用于存储像素数组，所以我认为一个好的估计是像素宽度乘以像素高度乘以每个通道中的位数乘以通道数。从MSDN 看来，每个通道有 32 位（4 字节）和 4 个通道。

我使用类似于以下的代码进行了一些测试，得到了非常好的结果：

var processed = 
    Enumerable
    .Range(0, 100)
    .Select(_ => new WriteableBitmap(
        800, 
        600, 
        96, 
        96, 
        PixelFormats.Bgr24, 
        new BitmapPalette(new List<Color>() { new Color() })))
    .Select(x => new { Bitmap = x, ByteSize = x.PixelWidth * x.PixelHeight * 4 * 4)
    .ToObservable()
    .Do(x => GC.AddMemoryPressure(x.ByteSize))
    .SelectMany(x => DoSomethingAsync(x.Bitmap));

processed
.Subscribe(x => GC.RemoveMemoryPressure(x.ByteSize));

但是，如果您的来源发布位图的速度比您处理它们的速度快，那么您仍然会遇到问题。背压会导致分配内存的速度快于释放内存的速度。

老实说，您真的有位图推送给您吗？我不知道您的实际程序是什么样的，但在您的示例代码中，这显然是一个基于拉取的系统。如果它是基于拉的系统，您是否考虑过PLINQ？ PLINQ 非常适合这种类型的事情。它使您可以很好地控制并发性，并且您不必担心背压。

【讨论】：

Do 语句的出色（正确）用法。很高兴看到一个不只是记录的示例。
@LeeCampbell - 来自写这本书的人的道具。太棒了！

【解决方案2】：

在我看来，您遇到了一个简单的内存使用问题。

如果每个通道有 4 个字节，每个像素有 4 个通道，那么您的 1000 张 800 x 600 图像每张为 1000 x 800 x 600 x 4 x 4 = 733MB（大约）。

然而，令我印象深刻的是，在您的代码中可能让您感到悲痛的是，您从一个可枚举的对象开始，然后将其转换为一个可观察的对象，该可观察对象是使用任务构建的，最终您以异步方式运行用一把火忘记.Subscribe()，然后你用await Task.Delay(20000); 捏造回报。这一切都容易出错。你应该避免混合你的“monads”。

我会这样写：

public async Task BitmapPipelineTest()
{
    await
        Observable
            .Range(0, 100)
            .Select(_ => new WriteableBitmap(
                800, 600, 96, 96,
                PixelFormats.Bgr24,
                new BitmapPalette(new List<Color>() { new Color() })))
            .SelectMany(x =>
                Observable
                    .Start(() =>
                    {
                        Thread.Sleep(10);
                        return new object();
                    }));
}

【讨论】：