c# - 在不同线程上写入多个表

Question

我正在使用一个包含很多列的大型 csv 文件。我想将该文件拆分为 n 个不同类型的对象，然后将这些不同的对象批量插入到 n 个不同的表中。我想尽可能地优化，而不是在写入数据库时​​将整个 csv 文件保存在内存中。如果写入 1 个表时出错，则所有表都必须回滚它们的事务。我正在传递 token 源，因为如果 1 个表出现问题，我希望所有其他表停止处理和回滚。

为了简化对象的创建/定义，我使用了 ExpandoObjects。

TableDetails 类包含表名、SqlConnection 和将用于写入表的事务。

我包含了实现 IDataReader 的 ObjectDataReader，以便于通过 SqlBulkCopy 发送 IEnumerable。

为什么当我运行虚拟 SaveToDb 方法时，所有 3 个表都有自己的线程来创建和写入控制台，但当我运行真正的 SaveToDb 方法时，所有工作都在 1 个线程上完成？

我需要做什么才能获得与我的测试方法相同的行为？

创建的虚拟表如下

create table Table1(Column1 int)
create table Table2(Column1 int)
create table Table3(Column1 int)

这里的主要工作

    void Main()
    {
        var tokenSource = new CancellationTokenSource();
        
        var sqlConnection1 = new SqlConnection("Some Connection String");
        sqlConnection1.Open();
        var sqlConnection2= new SqlConnection("Some Connection String");
        sqlConnection2.Open();
        var sqlConnection3 = new SqlConnection("Some Connection String");
        sqlConnection3.Open();
        
        var details = new List<TableDetails>()
        {
            new TableDetails(){ TableName = "Table1", Connection = sqlConnection1, Transaction = sqlConnection1.BeginTransaction(), ColumnMap = new Dictionary<int,string>(){{0, "Colunm1"}}},
            new TableDetails(){ TableName = "Table2", Connection = sqlConnection2, Transaction = sqlConnection2.BeginTransaction(), ColumnMap = new Dictionary<int,string>(){{1, "Colunm1"}}},
            new TableDetails(){ TableName = "Table3",  Connection = sqlConnection3, Transaction = sqlConnection3.BeginTransaction(),ColumnMap = new Dictionary<int,string>(){{2, "Colunm1"}}},
        };
        
        var lines = GetLines(100);
    
        var tasks = lines
            .SelectMany(e => SplitUp(e, details))
            .GroupBy(e => e.Item1)
            .Select(e => new { e.Key, Value = e.Select(v => MakeExpando(v.Item2, v.Item1)) })
    //      .Select(e => SaveToDbTest(e.Key, e.Value));
            .Select(e => SaveToDb(e.Value, e.Key, tokenSource));
    
        Task.WhenAll(tasks).Wait();
        
        foreach (var detail in details)
        {
            detail.Transaction.Commit();
            detail.Connection.Dispose();
        }
    }
    
    public IEnumerable<string> GetLines(int size)
    {
        var rand = new Random();
        for (int i = 0; i < size; i++)
            yield return $"{rand.Next(1, 100)},{rand.Next(1, 100)},{rand.Next(1, 100)}";
    }
    
    public IEnumerable<(TableDetails, string)> SplitUp(string line, List<TableDetails> details)
    {
        foreach (var detail in details)
        {
            yield return (detail, line);
        }
    }
    
    public ExpandoObject MakeExpando(string line, TableDetails details)
    {
        Console.WriteLine($"Thread ID:{Thread.CurrentThread.ManagedThreadId} Making Expando for Table {details.TableName}");
        
        var items = line.Split(',');
        dynamic retVal = new ExpandoObject();
        var r = retVal as IDictionary<string, object>;
        
        object value;
        foreach(var map in details.ColumnMap)
        {
            value = items[map.Key];
            r.Add(map.Value, value);
        }
        
        return retVal;
    }
    
    public Task SaveToDbTest(TableDetails details, IEnumerable<ExpandoObject> items)
    {
        var retVal = Task.Factory.StartNew(() =>
         {
             foreach (var i in items)
             {
                 Console.WriteLine($"Thread ID:{Thread.CurrentThread.ManagedThreadId} Saving To Table {details.TableName} => {i}");
             }
         });
    
        return retVal;
    }
    
    private async Task SaveToDb<T>(IEnumerable<T> items, TableDetails details, CancellationTokenSource tokenSource) where T : IDictionary<string, object>
    {
        var bulkCopy = new SqlBulkCopy(details.Connection, SqlBulkCopyOptions.Default, details.Transaction);
    
        try
        {
            bulkCopy.DestinationTableName = details.TableName;
            bulkCopy.BatchSize = 20;
            bulkCopy.BulkCopyTimeout = (int)TimeSpan.FromMinutes(120).TotalSeconds;
            bulkCopy.EnableStreaming = true;
    
            var reader = new ObjectDataReader<T>(items, details.ColumnMap.Count());
    
            var stopwatch = new Stopwatch();
            stopwatch.Start();
    
            await bulkCopy.WriteToServerAsync(reader, tokenSource.Token);
            stopwatch.Stop();
            Console.WriteLine($"completed db write in {stopwatch.Elapsed}");
        }
        catch (Exception ex)
        {
            if (ex.GetType() != typeof(TaskCanceledException))
                tokenSource.Cancel();
            throw;
        }
    }

表格详细信息

    public class TableDetails
    {
        public string TableName { get; set; }
        public SqlConnection Connection { get; set; }
        public SqlTransaction Transaction { get; set; }
        public Dictionary<int, string> ColumnMap {get; set;}
    }

还有一个 IDataReader

    public class ObjectDataReader<TData> : IDataReader where TData : IDictionary<string, object>
    {
        private IEnumerator<TData> _dataEnumerator;
        private Dictionary<int, string> _indexToName;
    
    
        public ObjectDataReader(IEnumerable<TData> data, int propertyCount)
        {
            _fieldCount = propertyCount;
            _dataEnumerator = data.GetEnumerator();
        }
    
        #region IDataReader Members
    
        public void Close()
        {
            Dispose();
        }
    
        public int Depth => 1;
    
        public DataTable GetSchemaTable()
        {
            return null;
        }
    
        public bool IsClosed => _dataEnumerator == null;
    
        public bool NextResult()
        {
            return false;
        }
    
        public bool Read()
        {
            if (IsClosed)
                throw new ObjectDisposedException(GetType().Name);
            Console.WriteLine($"Thread ID:{Thread.CurrentThread.ManagedThreadId} Reading next item");
            return _dataEnumerator.MoveNext();
        }
    
        public int RecordsAffected => -1;
    
        #endregion
    
        #region IDisposable Members
    
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }
    
        private void Dispose(bool disposing)
        {
            if (disposing)
            {
                if (_dataEnumerator != null)
                {
                    _dataEnumerator.Dispose();
                    _dataEnumerator = null;
                }
            }
        }
    
        #endregion
    
        #region IDataRecord Members
    
        public int GetOrdinal(string name)
        {
            throw new NotImplementedException();
        }
    
        public object GetValue(int i)
        {
            if (_dataEnumerator == null)
                throw new ObjectDisposedException(GetType().Name);
    
            var item = _dataEnumerator.Current;
    
            if (_indexToName == null)
            {
                _indexToName = item
                    .Select((e, id) => new { Index = id, e.Key })
                    .ToDictionary(k => k.Index, v => v.Key);
            }
    
            if (_indexToName.Count <= i)
                return null;
    
            return item[_indexToName[i]];
        }
    
        private int _fieldCount;
        public int FieldCount => _fieldCount; //throw new NotImplementedException(); // s_propertyAccessorCache.Value.Accessors.Count;
    
        #region Not Implemented Members
    
        public bool GetBoolean(int i)
        {
            throw new NotImplementedException();
        }
    
        public byte GetByte(int i)
        {
            throw new NotImplementedException();
        }
    
        public long GetBytes(int i, long fieldOffset, byte[] buffer, int bufferoffset, int length)
        {
            throw new NotImplementedException();
        }
    
        public char GetChar(int i)
        {
            throw new NotImplementedException();
        }
    
        public long GetChars(int i, long fieldoffset, char[] buffer, int bufferoffset, int length)
        {
            throw new NotImplementedException();
        }
    
        public IDataReader GetData(int i)
        {
            throw new NotImplementedException();
        }
    
        public string GetDataTypeName(int i)
        {
            throw new NotImplementedException();
        }
    
        public DateTime GetDateTime(int i)
        {
            throw new NotImplementedException();
        }
    
        public decimal GetDecimal(int i)
        {
            throw new NotImplementedException();
        }
    
        public double GetDouble(int i)
        {
            throw new NotImplementedException();
        }
    
        public Type GetFieldType(int i)
        {
            throw new NotImplementedException();
        }
    
        public float GetFloat(int i)
        {
            throw new NotImplementedException();
        }
    
        public Guid GetGuid(int i)
        {
            throw new NotImplementedException();
        }
    
        public short GetInt16(int i)
        {
            throw new NotImplementedException();
        }
    
        public int GetInt32(int i)
        {
            throw new NotImplementedException();
        }
    
        public long GetInt64(int i)
        {
            throw new NotImplementedException();
        }
    
        public string GetName(int i)
        {
            throw new NotImplementedException();
        }
    
        public string GetString(int i)
        {
            throw new NotImplementedException();
        }
    
        public int GetValues(object[] values)
        {
            throw new NotImplementedException();
        }
    
        public bool IsDBNull(int i)
        {
            var val = GetValue(i);
            return val == null;
            throw new NotImplementedException();
        }
    
        public object this[string name]
        {
            get { throw new NotImplementedException(); }
        }
    
        public object this[int i]
        {
            get { throw new NotImplementedException(); }
        }
    
        #endregion
    
        #endregion
    }

写入数据库时​​的输出。

    Thread ID:60 Reading next item
    Thread ID:60 Making Expando for Table Table1
    Thread ID:60 Reading next item
    Thread ID:60 Making Expando for Table Table1
    ...
    Thread ID:60 Reading next item
    Thread ID:60 Making Expando for Table Table2
    Thread ID:60 Reading next item
    ...
    Thread ID:60 Making Expando for Table Table3
    Thread ID:60 Reading next item
    Thread ID:60 Making Expando for Table Table3
    Thread ID:60 Reading next item

写入控制台时的输出。

    Thread ID:62 Making Expando for Table Table2
    Thread ID:71 Making Expando for Table Table3
    Thread ID:69 Making Expando for Table Table1
    Thread ID:62 Saving To Table Table2 => System.Dynamic.ExpandoObject
    Thread ID:62 Making Expando for Table Table2
    ...
    Thread ID:71 Saving To Table Table3 => System.Dynamic.ExpandoObject
    Thread ID:71 Making Expando for Table Table3
    ...
    Thread ID:62 Making Expando for Table Table2
    Thread ID:62 Saving To Table Table2 => System.Dynamic.ExpandoObject
    Thread ID:62 Making Expando for Table Table2
    Thread ID:62 Saving To Table Table2 => System.Dynamic.ExpandoObject
    Thread ID:62 Making Expando for Table Table2

Answer 1

一些大学向我指出，我的 SaveToDBTest 方法和我的 SaveToDB 方法之间存在差异。即 SaveToDbTest 创建了一个新任务，并启动了它。 Task.WhenAll() 逐个枚举并启动任务。

以下代码更改使所有内容都在单独的线程上运行。

    var tasks = lines
        .SelectMany(e => SplitUp(e, details))
        .GroupBy(e => e.Item1)
        .Select(e => new { e.Key, Value = e.Select(v => MakeExpando(v.Item2, v.Item1)) })
        .Select(e => Task.Run(() => SaveToDb(e.Value, e.Key, tokenSource)));