具有并发队列的 sqlbulkcopy答案

【问题标题】：sqlbulkcopy with concurrent queue具有并发队列的 sqlbulkcopy
【发布时间】：2017-01-13 14:18:45
【问题描述】：

我有一个使用类对象定义的并发队列，其中包含如下定义的 65000 条记录

ConcurrentQueue<Data> DataQueue = new ConcurrentQueue<Data>();

public class Data
{
    public string Id { get; set; }
    public string T { get; set; }
    public string D { get; set; }
    public string L { get; set; }
    public string I { get; set; }
    public string V { get; set; }
}

我正在使用以下代码插入数据库

public void InsertIntoDB()
    {

        using (cn = new SqlConnection(connectionString))
        {
            cn.Open();

            Data item;
            while (SpotDataQueue.Count > 0)
            {
                if (DataQueue.TryDequeue(out item))
                {
                    using (SqlCommand cm = cn.CreateCommand())
                    {
                        cm.CommandText = @"INSERT INTO [TableName] ([WId], [L], [I], [V],[JId],[I],[DateTime]) VALUES (@WId, @l, @i, @v, @jid,@i,@dt)";
                        cm.Parameters.AddWithValue("WId", item.Id);
                        cm.Parameters.AddWithValue("@l", item.L);
                        cm.Parameters.AddWithValue("@i", item.I);
                        cm.Parameters.AddWithValue("@v", item.V);
                        cm.Parameters.AddWithValue("@jid", 1);
                        cm.Parameters.AddWithValue("@i", false);
                        cm.Parameters.AddWithValue("@dt", DateTime.Now);
                        cm.ExecuteNonQuery();
                    }
                }
            }
        }
    }

表结构：

WId         nvarchar(50)    AllowNulls
L           nvarchar(MAX)   AllowNulls
I           nvarchar(MAX)   AllowNulls
V           nvarchar(MAX)   AllowNulls
JId         int             AllowNulls
I           bit             AllowNulls
DateTime    datetime        AllowNulls

如何将 Data 类型的并发队列转换为 DATATABLE 或 DATAREADER 以使 SQLBULKCOPY 成为可能？

谢谢。

【问题讨论】：

我会小心你的InsertIntoDB 函数，如果你插入的速度比你写入并发队列的速度快，你的while 循环将退出，之后将添加新数据。考虑切换到 new BlockingCollection<Data>(new ConcurrentQueue<Data>()) 并使用 foreach(Data item in SpotDataQueue.GetConsumingEnumerable() 代替 while 循环和 TryDequeue
@ScottChamberlain ：嗨，队列已经加载了 65000 条记录，因此在这种情况下不存在这个问题。我也会考虑你的建议。谢谢:)

标签： c# sql-server-2012

【解决方案1】：

Here 是一种借助反射完成工作的方法（更改参数类型）：

public static DataTable ToDataTable<T>(this IEnumerable<T> items)
{
    var tb = new DataTable(typeof(T).Name);

    PropertyInfo[] props = typeof(T).GetProperties(BindingFlags.Public | BindingFlags.Instance);

    foreach(var prop in props)
    {
        tb.Columns.Add(prop.Name, prop.PropertyType);
    }

    foreach (var item in items)
    {
       var values = new object[props.Length];
        for (var i=0; i<props.Length; i++)
        {
            values[i] = props[i].GetValue(item, null);
        }

        tb.Rows.Add(values);
    }

    return tb;
}

现在您可以将它与您的 Queue 一起使用：

DataTable dataSourceForSqlBulkCopy = DataQueue.ToDataTable();

【讨论】：

您好，感谢您的回复，我会试试这个并告诉您:)