在自定义流类中调用其他 *Async 方法

Question

我有一个派生自 Stream 的类，在它的读/写中，我需要访问一个数据库。我想使用 SqlData 对象的 *Async 方法，但当然它希望我将 Read 的签名更改为具有 async 修饰符，如下所示（我认为）：

public async override Task <int> Read( byte[] buffer, int offset, int count )

这与实际签名不匹配，所以我收到一条编译错误消息，说明这一点。知道如何在此流中使用 *Async 还是应该让它保持同步？

我还使用了 https://blogs.msdn.microsoft.com/pfxteam/2011/01/15/asynclazyt/ 的 Stephen Toub 的 AsyncLazy

我的班级正在启用与 SQL Server 中 VarBinary(max) 字段之间的数据流，其中大部分想法来自 http://www.syntaxwarriors.com/2013/stream-varbinary-data-to-and-from-mssql-using-csharp/。

以下是代码中有趣的部分（已注释掉异步位）：

public class BinaryDataStream<T> : Stream
{
    /* Async */Lazy<SqlDataReader> lazyReader;
    SqlConnection connection;
    SqlCommand firstUpdate;
    SqlCommand otherUpdates;
    long position;

    public BinaryDataStream( DbContext context, string tableName, string keyName, string columnName, T keyValue )
    {
        connection = new SqlConnection( context.Database.GetDbConnection().ConnectionString );

        lazyReader = new /* Async */Lazy<SqlDataReader>( /* async */ () =>
        {
            using ( var cmd = new SqlCommand( $"SELECT TOP 1 [{columnName}] FROM [dbo].[{tableName}] WHERE [{keyName}] = @id", connection ) )
            {
                cmd.Parameters.AddWithValue( "@id", keyValue );
                /* await */ connection.Open/* Async */();

                var r = /* await */ cmd.ExecuteReader/* Async */( System.Data.CommandBehavior.SequentialAccess | System.Data.CommandBehavior.SingleResult | System.Data.CommandBehavior.SingleRow | System.Data.CommandBehavior.CloseConnection );
                r.Read();

                return r;
            }
        } );

        firstUpdate = new SqlCommand( $"UPDATE [dbo].[{tableName}] SET [{columnName}] = @firstchunk WHERE [{keyName}] = @id", connection );
        firstUpdate.Parameters.AddWithValue( "@id", keyValue );
        firstUpdate.Parameters.AddWithValue( "@firstchunk", new byte[] { } );

        otherUpdates = new SqlCommand( $"UPDATE [dbo].[{tableName}] SET [{columnName}].WRITE( @chunk, NULL, @length ) WHERE [{keyName}] = @id", connection );
        otherUpdates.Parameters.AddWithValue( "@id", keyValue );
        otherUpdates.Parameters.AddWithValue( "@length", 0 );
        otherUpdates.Parameters.AddWithValue( "@chunk", new byte[] { } );
    }

    public /* async */ override /* Task< */int/* > */ Read( byte[] buffer, int offset, int count )
    {
        var reader = /* await */ lazyReader.Value;
        var bytesRead = reader.GetBytes( 0, position, buffer, offset, count );
        position += bytesRead;
        return (int)bytesRead;
    }

    public /* async */ override void Write( byte[] buffer, int offset, int count )
    {
        if ( count == 0 ) return;

        /* await */ connection.Open/* Async */();

        try
        {
            if ( firstUpdate != null )
            {
                firstUpdate.Parameters[ "@firstchunk" ].Value = buffer;
                /* await */ firstUpdate.ExecuteNonQuery/* Async */();
                firstUpdate = null;
            }
            else
            {

                var chunk = buffer;
                if ( count < buffer.Length )
                {
                    chunk = new byte[ count ];
                    Array.Copy( buffer, 0, chunk, 0, count );
                }

                otherUpdates.Parameters[ "@chunk" ].Value = chunk;
                otherUpdates.Parameters[ "@length" ].Value = count;
                /* await */ otherUpdates.ExecuteNonQuery/* Async */();
            }
        }
        finally
        {
            connection.Close();
        }
    }
}

Answer 1

Stream 类具有其方法的同步和异步版本。同样，您正在执行的数据库操作也会公开其方法的同步和异步版本。用同步实现覆盖同步方法，用异步实现覆盖异步方法。