主页 / coding / 问题 / 79221507
Accepted
Josh
Asked: 2024-11-25 10:55:38 +0800 CST

如何有效地反序列化压缩的对象列表，将其添加到其中，然后再次压缩而不占用太多内存

  • 772

我继承了一些执行以下步骤的代码：

  1. 从压缩数据的字节数组开始，对其进行流式传输和解压缩
  2. 将其反序列化为对象列表
  3. 添加至列表
  4. 序列化列表
  5. 将数据重新压缩为字节数组
private static readonly RecyclableMemoryStreamManager MemoryStreamManager = new RecyclableMemoryStreamManager();

static async Task Main(string[] args)
{
    using (FileStream fileStream = new FileStream("C:\\data.txt", FileMode.Open, FileAccess.Read))
    {
        var dataList = await DecompressData(fileStream);
        dataList.Add(new MyObject { });
        using(var stream = MemoryStreamManager.GetStream())
        {
            await JsonSerializer.SerializeAsync(stream, dataList);
            stream.Position = 0;
            var b = await CompressData(stream);
        }
    }

    Console.WriteLine("All done");
}

private static async Task<List<SipTraceRecord>> DecompressData(Stream data)
{
    using (var resultStream = MemoryStreamManager.GetStream())
    {
        GZipStream gzip = new GZipStream(data, CompressionMode.Decompress);

        List<SipTraceRecord> recordsList = await JsonSerializer.DeserializeAsync<List<MyObject>>(gzip);
        return recordsList;
    }
}

private static async Task<byte[]> CompressData(Stream data)
{
    byte[] compressedData;
    using (var ms = MemoryStreamManager.GetStream())
    {
        using (GZipStream gzip = new GZipStream(ms, CompressionMode.Compress))
        {
            data.CopyTo(gzip);
            compressedData = ms.GetBuffer();
        }

    }

    return compressedData;
}

这与代码开始时有很大不同，我只是试图尽可能地优化内存。测试输入文件为 600Kb，解压后为 22Mb，之前它使用 100Mb 的内存。现在内存已降至 90Mb。内存使用率高的区域仍然存在，例如await JsonSerializer.SerializeAsync(stream, dataList);使用 10Mb 将数据写入流。这是否可以像在其他方向上一样进行优化，其中没有字节数组，只是根据需要进行流式传输？

这data.CopyTo(gzip);也是重复的数据，但此时数据已被压缩，因此仅使用<1Mb

c#

1 个回答

  1. Best Answer

    您无需反序列化和重新序列化列表，而是可以JsonSerializer.DeserializeAsyncEnumerable()分块流式传输输入数据，动态解压缩，然后动态压缩到包含 JSON 数组开头的某个输出流。之后，您可以流式传输和压缩新值，并将它们添加到 JSON 数组中。

    实现该操作的方法如下：

    public static class JsonExtensions
{
    // nugets required: 
    // System.Linq.Async, https://www.nuget.org/packages/System.Linq.Async/
    // Microsoft.IO.RecyclableMemoryStream https://www.nuget.org/packages/microsoft.io.recyclablememorystream/
    public static readonly RecyclableMemoryStreamManager MemoryStreamManager = new RecyclableMemoryStreamManager();
    
    // 2x buffer sized as recommended by Bradley Grainger, https://faithlife.codes/blog/2012/06/always-wrap-gzipstream-with-bufferedstream/
    // But anything smaller than 85,000 bytes should be OK, since objects larger than that go on the large object heap.  See:
    // https://learn.microsoft.com/en-us/dotnet/standard/garbage-collection/large-object-heap
    const int BufferSize = 16384;       

    // Compressed copy + serialize
    public static async Task<byte []> CopyAndAddToCompressedByteArrayAsync<TItem>(byte [] input, IEnumerable<TItem> newItems, JsonSerializerOptions? options = default)
    {
        using var inputStream = new MemoryStream(input);
        using var outputStream = MemoryStreamManager.GetStream();
        await CopyAndAddToCompressedStreamAsync(inputStream, outputStream, newItems, options);
        return outputStream.ToArray();
    }
    
    public static async Task CopyAndAddToCompressedFileAsync<TItem>(string inputPath, string outputPath, IEnumerable<TItem> newItems, JsonSerializerOptions? options = default)
    {
        await using var input = File.OpenRead(inputPath);
        await using var output = File.OpenWrite(outputPath);
        await CopyAndAddToCompressedStreamAsync(input, output, newItems, options);
    }
    
    public static async Task CopyAndAddToCompressedStreamAsync<TItem>(Stream input, Stream output, IEnumerable<TItem> newItems, JsonSerializerOptions? options = default)
    {
        options ??= JsonSerializerOptions.Default;
        
        await using var inputDecompressor = new GZipStream(input, CompressionMode.Decompress, leaveOpen : true);
        await using var outputCompressor = new GZipStream(output, CompressionMode.Compress, leaveOpen : true);
        await using var outputBuffer = new BufferedStream(outputCompressor, BufferSize);

        // ToAsyncEnumerable() and  oldItemsAsync!.Concat(newItemsAsync) are from System.Linq.Async, https://www.nuget.org/packages/System.Linq.Async/
        var oldItemsAsync = JsonSerializer.DeserializeAsyncEnumerable<TItem>(inputDecompressor, options) ?? Enumerable.Empty<TItem>().ToAsyncEnumerable();
        var allItemsAsync = oldItemsAsync!.Concat(newItems.ToAsyncEnumerable());
        
        // https://learn.microsoft.com/en-us/dotnet/core/compatibility/serialization/6.0/iasyncenumerable-serialization
        await JsonSerializer.SerializeAsync(outputBuffer, allItemsAsync, options);
    }

    // Compressed serialize (initial creation)
    public static async Task<byte []> SerializeToCompressedByteArrayAsync<TValue>(TValue value, JsonSerializerOptions? options = default)
    {
        using var output = JsonExtensions.MemoryStreamManager.GetStream();
        await JsonExtensions.SerializeToCompressedStreamAsync(output, value, options);
        return output.ToArray();
    }

    public static async Task SerializeToCompressedFileAsync<TValue>(string path, TValue value, JsonSerializerOptions? options = default)
    {
        await using var output = File.OpenWrite(path);
        await SerializeToCompressedStreamAsync(output, value, options);
    }
    
    public static async Task SerializeToCompressedStreamAsync<TValue>(Stream utf8Json, TValue value, JsonSerializerOptions? options = default)
    {
        await using var outputCompressor = new GZipStream(utf8Json, CompressionMode.Compress, leaveOpen : true);
        await using var outputBuffer = new BufferedStream(outputCompressor, BufferSize);
        await JsonSerializer.SerializeAsync<TValue>(outputBuffer, value, options);
    }
}

    现在，如果您完全使用文件（这是我的建议），您可以按如下方式创建初始 JSON 文件：

    List<SipTraceRecord> initialList = /*Initialize this however you do currently */ ;

var options = new JsonSerializerOptions
{
    // Add whatever you need here
};
await JsonExtensions.SerializeToCompressedFileAsync(filePath, initialList, options);

    要附加到文件，您可以执行以下操作：

    List<SipTraceRecord> addList = /*Initialize this however you do currently */;

var tempPath = Path.GetTempFileName();
await JsonExtensions.CopyAndAddToCompressedFileAsync(filePath, tempPath, addList, options);
File.Move(tempPath, filePath, true);

    或者，如果您确实需要使用内存中的字节数组来存储压缩数据，则可以按如下方式创建初始数组：

    var options = new JsonSerializerOptions
{
    // Add whatever you need here
};
var initialBytes = await JsonExtensions.SerializeToCompressedByteArrayAsync(initialList, options);

    并创建一个连接数组，如下所示：

    var appendedBytes = await JsonExtensions.CopyAndAddToCompressedByteArrayAsync(initialBytes, list, options);

    笔记：

    • 即使您使用DeserializeAsyncEnumerable()来流式传输MemoryStream，也必须以异步方式执行，因为没有易于使用的 API 来通过 System.Text.Json 同步流式传输 JSON 数组。

    • DeserializeAsyncEnumerable()将尝试从大小等于JsonSerializerOptions.DefaultBufferSize默认值 16,384 字节的流中读取一个字节块，反序列化该块中的所有数组项，然后一次性返回它们。这可以防止在流过大型数组时内存无限增长。

    • 在.NET 9中，System.Text.Json 添加了对NDJSON（换行符分隔的 JSON格式）的支持，该格式由一系列没有外部数组括号的连接 JSON 对象组成。如果您能够迁移到 .NET 9，那么切换到该格式可能会更容易，因为您可以将SipTraceRecord记录附加到文件末尾，而无需通过流式传输来查找数组末尾。

      有关详细信息，请参阅在 .NET 中解析不带括号（没有根对象）的 JSON 对象序列文本？ 。

    • 调用MemoryStream.ToArray()或RecyclableMemoryStream.ToArray()效率低下，因为它们总是会返回一个新数组，可能大到足以放入大对象堆中。由于您会遇到内存不足错误，因此这是不可取的，但由于您的 API 适用于byte[]数组，因此很难避免这种情况。您可以考虑将 API 更改为直接处理文件或RecyclableMemoryStream对象，而不是字节数组。

    • .NET 6IAsyncEnumerable<T>中已添加对 System.Text.Json的支持。此方法在早期版本中不起作用。

    此处的演示小提琴：.NET 9，.NET 8。

    • 6

相关问题