如何在不创建大缓冲区的情况下将.NET对象的大图形序列化为SQL Server BLOB？

Question

我们的代码如下:

ms = New IO.MemoryStream
bin = New System.Runtime.Serialization.Formatters.Binary.BinaryFormatter
bin.Serialize(ms, largeGraphOfObjects)
dataToSaveToDatabase = ms.ToArray()
// put dataToSaveToDatabase in a Sql server BLOB

但是内存蒸汽从大内存堆中分配了一个大缓冲区,这给我们带来了麻烦.那么我们如何在不需要足够的可用内存来保存序列化对象的情况下流式传输数据.

我正在寻找一种从SQL服务器获取Stream的方法,然后可以将其传递给bin.Serialize(),以避免将所有数据保存在我的进程内存中.

同样,为了阅读数据...

---

更多背景.

这是一个复杂的数字处理系统的一部分,它可以近乎实时地处理数据,寻找设备问题等,进行序列化以便在数据馈送等数据质量出现问题时重新启动(我们存储数据源)并且可以在运算符编辑出错误值后重新运行它们.)

因此,我们更频繁地序列化对象,然后我们对它们进行反序列化.

我们正在序列化的对象包括非常大的数组,这些数组主要是双精度数以及许多小的"更正常"的对象.我们正在推动32位系统的内存限制,并使车库收集器非常努力.(系统中的其他地方正在进行改进,例如重用大型数组而不是创建新数组.)

通常,状态的序列化是追踪内存异常的最后一根 ; 我们的峰值内存使用量是在进行此序列化时.

我认为当我们反序列化对象时会得到大的内存池碎片,我预计在给定数组大小的情况下,还存在大内存池碎片的其他问题.(这还没有被调查过,因为首先看过这个的人是数字处理专家,而不是内存管理专家.)

客户是否使用Sql Server 2000,2005和2008的混合,如果可能的话,我们宁愿不为每个版本的Sql Server使用不同的代码路径.

我们可以一次拥有许多活动模型(在不同的过程中,在许多机器上),每个模型可以有许多已保存的状态.因此,保存的状态存储在数据库blob中,而不是文件中.

由于保存状态的传播很重要,我宁愿不将对象序列化为文件,然后一次将文件放在BLOB中.

我问过的其他相关问题

• 如何从/到SQL Server BLOB字段流数据？
• 是否有类似SqlFileStream的类与Sql Server 2005一起使用？

Answer 1

没有内置的ADO.Net功能可以很好地处理大数据.问题有两个:

• 没有API可以将"写入"SQL命令或参数"写入"流.接受流的参数类型(如FileStream)接受来自它的READ流,这与写入流的序列化语义不一致.无论你转向哪种方式,最终都会得到整个序列化对象的内存副本,不好.
• 即使上面的点已经解决(并且它不可能),TDS协议和SQL Server接受参数的方式也不适用于大参数,因为在启动执行之前必须首先接收整个请求,这样就可以了在SQL Server中创建对象的其他副本.

所以你真的必须从不同的角度来看待它.幸运的是,有一个相当简单的解决方案.诀窍是使用高效UPDATE .WRITE语法并在一系列T-SQL语句中逐个传递数据块.这是MSDN推荐的方法,请参阅在ADO.NET中修改大值(最大)数据.这看起来很复杂,但实际上很容易做到并插入Stream类.

BlobStream类

这是解决方案的基础.Stream派生类,它实现Write方法作为对T-SQL BLOB WRITE语法的调用.直截了当,唯一有趣的是它必须跟踪第一次更新,因为UPDATE ... SET blob.WRITE(...)语法在NULL字段上会失败:

class BlobStream: Stream
{
    private SqlCommand cmdAppendChunk;
    private SqlCommand cmdFirstChunk;
    private SqlConnection connection;
    private SqlTransaction transaction;

    private SqlParameter paramChunk;
    private SqlParameter paramLength;

    private long offset;

    public BlobStream(
        SqlConnection connection,
        SqlTransaction transaction,
        string schemaName,
        string tableName,
        string blobColumn,
        string keyColumn,
        object keyValue)
    {
        this.transaction = transaction;
        this.connection = connection;
        cmdFirstChunk = new SqlCommand(String.Format(@"
UPDATE [{0}].[{1}]
    SET [{2}] = @firstChunk
    WHERE [{3}] = @key"
            ,schemaName, tableName, blobColumn, keyColumn)
            , connection, transaction);
        cmdFirstChunk.Parameters.AddWithValue("@key", keyValue);
        cmdAppendChunk = new SqlCommand(String.Format(@"
UPDATE [{0}].[{1}]
    SET [{2}].WRITE(@chunk, NULL, NULL)
    WHERE [{3}] = @key"
            , schemaName, tableName, blobColumn, keyColumn)
            , connection, transaction);
        cmdAppendChunk.Parameters.AddWithValue("@key", keyValue);
        paramChunk = new SqlParameter("@chunk", SqlDbType.VarBinary, -1);
        cmdAppendChunk.Parameters.Add(paramChunk);
    }

    public override void Write(byte[] buffer, int index, int count)
    {
        byte[] bytesToWrite = buffer;
        if (index != 0 || count != buffer.Length)
        {
            bytesToWrite = new MemoryStream(buffer, index, count).ToArray();
        }
        if (offset == 0)
        {
            cmdFirstChunk.Parameters.AddWithValue("@firstChunk", bytesToWrite);
            cmdFirstChunk.ExecuteNonQuery();
            offset = count;
        }
        else
        {
            paramChunk.Value = bytesToWrite;
            cmdAppendChunk.ExecuteNonQuery();
            offset += count;
        }
    }

    // Rest of the abstract Stream implementation
 }

使用BlobStream

要使用这个新创建的blob流类,请插入一个BufferedStream.该类有一个简单的设计,只处理将流写入表的列.我将重用另一个示例中的表:

CREATE TABLE [dbo].[Uploads](
    [Id] [int] IDENTITY(1,1) NOT NULL,
    [FileName] [varchar](256) NULL,
    [ContentType] [varchar](256) NULL,
    [FileData] [varbinary](max) NULL)

我将添加一个要序列化的虚拟对象:

[Serializable]
class HugeSerialized
{
    public byte[] theBigArray { get; set; }
}

最后,实际序列化.我们首先在Uploads表中插入一条新记录,然后BlobStream在新插入的Id上创建一个并将序列化直接调用到此流中:

using (SqlConnection conn = new SqlConnection(Settings.Default.connString))
{
    conn.Open();
    using (SqlTransaction trn = conn.BeginTransaction())
    {
        SqlCommand cmdInsert = new SqlCommand(
@"INSERT INTO dbo.Uploads (FileName, ContentType)
VALUES (@fileName, @contentType);
SET @id = SCOPE_IDENTITY();", conn, trn);
        cmdInsert.Parameters.AddWithValue("@fileName", "Demo");
        cmdInsert.Parameters.AddWithValue("@contentType", "application/octet-stream");
        SqlParameter paramId = new SqlParameter("@id", SqlDbType.Int);
        paramId.Direction = ParameterDirection.Output;
        cmdInsert.Parameters.Add(paramId);
        cmdInsert.ExecuteNonQuery();

        BlobStream blob = new BlobStream(
            conn, trn, "dbo", "Uploads", "FileData", "Id", paramId.Value);
        BufferedStream bufferedBlob = new BufferedStream(blob, 8040);

        HugeSerialized big = new HugeSerialized { theBigArray = new byte[1024 * 1024] };
        BinaryFormatter bf = new BinaryFormatter();
        bf.Serialize(bufferedBlob, big);

        trn.Commit();
    }
}

如果您监视此简单示例的执行,您将看到无处创建大型序列化流.该示例将分配[1024*1024]的数组,但这是为了演示目的,有序列化的东西.此代码以缓冲方式序列化,按块进行块化,一次使用SQL Server BLOB建议的更新大小8040字节.

Answer 2

您只需要.NET Framework 4.5和流媒体.假设我们在硬盘上有一个大文件,我们想上传这个文件.

SQL代码:

CREATE TABLE BigFiles 
(
    [BigDataID] [int] IDENTITY(1,1) NOT NULL,
    [Data] VARBINARY(MAX) NULL
)

C#代码:

using (FileStream sourceStream = new FileStream(filePath, FileMode.Open))
{
    using (SqlCommand cmd = new SqlCommand(string.Format("UPDATE BigFiles SET Data=@Data WHERE BigDataID = @BigDataID"), _sqlConn))
    {
        cmd.Parameters.AddWithValue("@Data", sourceStream);
        cmd.Parameters.AddWithValue("@BigDataID", entryId);

        cmd.ExecuteNonQuery();
    }
}

对我有用.我已经成功上传了400 MB的文件,而当我尝试将此文件加载到内存时,MemoryStream抛出异常.

UPD:此代码适用于Windows 7,但在Windows XP和2003 Server上失败.