c# - 哪种方法限制大 blob 的内存使用 : SqlReader. GetBytes 或 SqlReader.GetStream？

Question

我想确定如何限制从本地数据库检索 blob 并通过 block 将其传输到第三方 Web 服务的作业中的内存使用。

使用 SqlDataReader，我似乎有两个选择:

1. 创建一个方法，该方法使用带有偏移量的 GetBytes 来检索返回 byte[] 的 blob 的一部分。然后，该方法的调用者将负责发出网络请求以传输该 block 。
2. 创建一个使用 GetStream 的方法，并向 ReadAsync 发出多个请求以填充 byte[] 缓冲区，并使用此缓冲区发出 Web 请求，直到文档传输完毕。

我更喜欢选项 1，因为它限制了方法的责任，但是如果我调用带有偏移量的 GetBytes，它会将整个偏移量加载到内存中还是 sql server 能够只返回请求的小块？如果我使用选项 2，那么该方法将有两个职责，从数据库加载一个 block 并发出 Web 请求以将文档存储在别处。

// option 1
public async Task<Tuple<int, byte[]>> GetDocumentChunk(int documentId, int offset, int maxChunkSize)
{
    var buffer = new byte[maxChunkSize];

    string sql = "SELECT Data FROM Document WHERE Id = @Id";

    using (SqlConnection connection = new SqlConnection(ConnectionString))
    {
        await connection.OpenAsync();

        using (SqlCommand command = new SqlCommand(sql, connection))
        {
            command.Parameters.AddWithValue("@Id", documentId);

            using (SqlDataReader reader = await command.ExecuteReaderAsync(CommandBehavior.SequentialAccess))
            {
                if (await reader.ReadAsync())
                {
                    int bytesRead = (int)reader.GetBytes(0, offset, buffer, 0, maxChunkSize);
                    return new Tuple<int, byte[]>(bytesRead, buffer);
                }
            }
        }
    }

    return new Tuple<int, byte[]>(0, buffer);
}

//option 2
public async Task<CallResult> TransferDocument(int documentId, int maxChunkSize)
{
    var buffer = new byte[maxChunkSize];

    string sql = "SELECT Data FROM Document WHERE Id = @Id";

    using (SqlConnection connection = new SqlConnection(ConnectionString))
    {
        await connection.OpenAsync();

        using (SqlCommand command = new SqlCommand(sql, connection))
        {
            command.Parameters.AddWithValue("@Id", documentId);

            using (SqlDataReader reader = await command.ExecuteReaderAsync(CommandBehavior.SequentialAccess))
            {
                using (Stream uploadDataStream = reader.GetStream(0))
                {
                    CallResult callResult;
                    int bytesRead;
                    do
                    {
                        bytesRead = await uploadDataStream.ReadAsync(buffer, 0, maxChunkSize);
                        callResult = await MyWebRequest(documentId, buffer, bytesRead);
                        if (callResult != CallResult.Success)
                        {
                            return callResult;
                        }
                    } while (bytesRead > 0);

                    return callResult;
                }
            }
        }
    }
}

Answer 1

使用选项 1，您将向源发出许多请求以获取数据，并且 GetBytes 不会在 SQL Server 上“搜索”流(如果搜索到，我会感到惊讶)，即将是一个非常低效的解决方案。

IAsyncEnumerable

使用选项 2，您可以获取流并按需处理它，因此您将发出单个数据库请求并从异步 I/O 中获得所有好处。

使用 C# 8 IAsyncEnumerable将完美解决您的问题，但它目前处于 Preview 阶段。

CopyToAsync

如果您可以获得需要上传内容的流，那么您可以使用 CopyToAsync .但我假设每个 block 都将在单独的请求中上传。如果是这样，您可以引入一个组件，它会像 Stream 一样嘎嘎，但实际上会在 DB stream calls CopyToAsync() 时将内容上传到网站。在上面:

class WebSiteChunkUploader : Stream
{
     private HttpClient _client = new HttpClient();
     public override bool CanWrite => true;
     public override bool CanRead => false;

     public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) =>

         await _client.PostAsync("localhost", new ByteArrayContent(buffer,offset, count));
}

旧的好IEnumerable

不幸的是，您不能将 IEnumerable 的 yield return 与 async/await 混合使用。但是，如果您决定使用阻塞 api 读取流，例如 Read，那么您可以使用旧的良好 yield return 重写它:

public IEnumerable<Tuple<byte[],int>> TransferDocument(int documentId, int maxChunkSize)
{
    string sql = "SELECT Data FROM Document WHERE Id = @Id";
    var buffer = new byte[maxChunkSize];
    using (SqlConnection connection = new SqlConnection(ConnectionString))
    {
        connection.Open();
        using (SqlCommand command = new SqlCommand(sql, connection))
        {
            command.Parameters.AddWithValue("@Id", documentId);
            using (SqlDataReader reader = command.ExecuteReader(CommandBehavior.SequentialAccess))
            using (Stream uploadDataStream = reader.GetStream(0))
            {
                while(var bytesRead = uploadDataStream.Read(buffer, 0, maxChunkSize)) > 0)
                   yield return Tuple(buffer, bytesRead);
            }
        }
    }
}

...
async Task DoMyTransfer() 
{
  foreach(var buffer in TransferDocument(1, 10000)) {
    await moveBytes(buffer)
  }
}

在这种情况下，您不会有与数据库的异步 IO 和奇特的 Tasks，但我想您无论如何都需要限制此上传操作，以免数据库因连接而过载。