使用 SqlFileStream 从 WCF 服务返回流

Question

我有一个 WCF 服务，用户可以从中请求大型数据文件（存储在启用 FileStream 的 SQL 数据库中）。这些文件应该是流式传输的，而不是在发送之前加载到内存中。

所以我有以下方法应该返回一个由 WCF 服务调用的流，以便它可以将 Stream 返回给客户端。

public static Stream GetData(string tableName, string columnName, string primaryKeyName, Guid primaryKey)
    {
        string sqlQuery =
            String.Format(
                "SELECT {0}.PathName(), GET_FILESTREAM_TRANSACTION_CONTEXT() FROM {1} WHERE {2} = @primaryKey", columnName, tableName, primaryKeyName);

        SqlFileStream stream;

        using (TransactionScope transactionScope = new TransactionScope())
        {
            byte[] serverTransactionContext;
            string serverPath;
            using (SqlConnection sqlConnection = new SqlConnection(ConfigurationManager.ConnectionStrings["ConnString"].ToString()))
            {
                sqlConnection.Open();

                using (SqlCommand sqlCommand = new SqlCommand(sqlQuery, sqlConnection))
                {
                    sqlCommand.Parameters.Add("@primaryKey", SqlDbType.UniqueIdentifier).Value = primaryKey;

                    using (SqlDataReader sqlDataReader = sqlCommand.ExecuteReader())
                    {
                        sqlDataReader.Read();
                        serverPath = sqlDataReader.GetSqlString(0).Value;
                        serverTransactionContext = sqlDataReader.GetSqlBinary(1).Value;
                        sqlDataReader.Close();
                    }
                }
            }

            stream = new SqlFileStream(serverPath, serverTransactionContext, FileAccess.Read);
            transactionScope.Complete();
        }

        return stream;
    }

我的问题在于 TransactionScope 和 SqlConnection。我现在这样做的方式不起作用，我收到一个 TransactionAbortedException 说“事务已中止”。我可以在返回 Stream 之前关闭事务和连接吗？感谢您的帮助，谢谢

编辑：

我为 SqlFileStream 创建了一个包装器，它实现了 IDisposable，以便在释放流后我可以关闭所有内容。似乎工作正常

public class WcfStream : Stream
{
    private readonly SqlConnection sqlConnection;
    private readonly SqlDataReader sqlDataReader;
    private readonly SqlTransaction sqlTransaction;
    private readonly SqlFileStream sqlFileStream;

    public WcfStream(string connectionString, string columnName, string tableName, string primaryKeyName, Guid primaryKey)
    {
        string sqlQuery =
            String.Format(
                "SELECT {0}.PathName(), GET_FILESTREAM_TRANSACTION_CONTEXT() FROM {1} WHERE {2} = @primaryKey",
                columnName, tableName, primaryKeyName);

        sqlConnection = new SqlConnection(connectionString);
        sqlConnection.Open();

        sqlTransaction = sqlConnection.BeginTransaction();

        using (SqlCommand sqlCommand = new SqlCommand(sqlQuery, sqlConnection, sqlTransaction))
        {
            sqlCommand.Parameters.Add("@primaryKey", SqlDbType.UniqueIdentifier).Value = primaryKey;
            sqlDataReader = sqlCommand.ExecuteReader();
        }

        sqlDataReader.Read();

        string serverPath = sqlDataReader.GetSqlString(0).Value;
        byte[] serverTransactionContext = sqlDataReader.GetSqlBinary(1).Value;

        sqlFileStream = new SqlFileStream(serverPath, serverTransactionContext, FileAccess.Read);
    }

    protected override void Dispose(bool disposing)
    {
        sqlDataReader.Close();
        sqlFileStream.Close();
        sqlConnection.Close();
    }

    public override void Flush()
    {
        sqlFileStream.Flush();
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        return sqlFileStream.Seek(offset, origin);
    }

    public override void SetLength(long value)
    {
        sqlFileStream.SetLength(value);
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        return sqlFileStream.Read(buffer, offset, count);
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        sqlFileStream.Write(buffer, offset, count);
    }

    public override bool CanRead
    {
        get { return sqlFileStream.CanRead; }
    }

    public override bool CanSeek
    {
        get { return sqlFileStream.CanSeek; }
    }

    public override bool CanWrite
    {
        get { return sqlFileStream.CanWrite; }
    }

    public override long Length
    {
        get { return sqlFileStream.Length; }
    }

    public override long Position
    {
        get { return sqlFileStream.Position; }
        set { sqlFileStream.Position = value; }
    }
}

Answer 1

通常我可能会建议将流包装在自定义流中，该流在处理时关闭事务，但是 IIRC WCF 不保证哪些线程做什么，但 TransactionScope 是特定于线程的。因此，也许更好的选择是将数据复制到MemoryStream（如果它不是太大的话）并返回它。 4.0 中的Stream.Copy 方法应该让这变得轻而易举，但请记住在最终的return (.Position = 0) 之前倒回内存流。

显然，如果流很大，这将是一个大问题，...但是，如果流足够大， 值得关注，那么个人我担心它在TransactionScope 中运行根本，因为它具有内置的时间限制，并导致可序列化的隔离（默认情况下）。

最后的建议是使用SqlTransaction，它不依赖于线程；您可以编写一个位于SqlFileStream 周围的Stream 包装器，然后关闭Dispose() 中的阅读器、事务和连接（以及包装的流）。 WCF 将在处理结果后调用它（通过Close()）。

Answer 2

嗯，我可能在这里遗漏了一些东西，但在我看来，更简单的方法是将流提供给 WCF 方法并从那里写入，而不是尝试返回客户端从中读取的流？

以下是 WCF 方法的示例：

public void WriteFileToStream(FetchFileArgs args, Stream outputStream)
{
    using (SqlConnection conn = CreateOpenConnection())
    using (SqlTransaction tran = conn.BeginTransaction(IsolationLevel.ReadCommitted))
    using (SqlCommand cmd = conn.CreateCommand())
    {
        cmd.CommandType = CommandType.StoredProcedure;
        cmd.CommandText = "usp_file";
        cmd.Transaction = tran;
        cmd.Parameters.Add("@FileId", SqlDbType.NVarChar).Value = args.Id;

        using (SqlDataReader reader = cmd.ExecuteReader())
        {
            if (reader.Read())
            {
                string path = reader.GetString(3);
                byte[] streamContext = reader.GetSqlBytes(4).Buffer;

                using (var sqlStream = new SqlFileStream(path, streamContext, FileAccess.Read))
                    sqlStream.CopyTo(outputStream);
            }
        }

        tran.Commit();
    }
}

在我的应用程序中，消费者恰好是一个 ASP.NET 应用程序，调用代码如下所示：

_fileStorageProvider.WriteFileToStream(fileId, Response.OutputStream);

Answer 3

从逻辑上讲，与 SQL 相关的东西都不属于 Stream wrapper 类 (WcfStream)，尤其是当您打算将 WcfStream 实例发送到外部客户端时。

您可以做的是，一旦 WcfStream 被释放或关闭，就会触发一个事件：

public class WcfStream : Stream
{
    public Stream SQLStream { get; set; }
    public event EventHandler StreamClosedEventHandler;

    protected override void Dispose(bool disposing)
    {
        if (disposing)
        {
            SQLStream.Dispose();

            if (this.StreamClosedEventHandler != null)
            {
                this.StreamClosedEventHandler(this, new EventArgs());
            }
        }
        base.Dispose(disposing);
    }
}

然后在您的主代码中，您将一个事件处理程序连接到 StreamClosedEventHandler 并关闭所有与 sql 相关的对象：

...
    WcfStream test = new WcfStream();
    test.SQLStream = new SqlFileStream(filePath, txContext, FileAccess.Read);
    test.StreamClosedEventHandler +=
                new EventHandler((sender, args) => DownloadStreamCompleted(sqlDataReader, sqlConnection));

    return test;
}

private void DownloadStreamCompleted(SqlDataReader sqlDataReader, SQLConnection sqlConnection)
{
    // You might want to commit Transaction here as well
    sqlDataReader.Close();
    sqlConnection.Close();
}

这看起来对我有用，它使流式处理逻辑与 SQL 相关代码分开。