如何在并行/多线程中使用 dapper 从 sql 数据库中获取数据？答案

【问题标题】：How to get data from sql database using dapper in parallel/multiple threads?如何在并行/多线程中使用 dapper 从 sql 数据库中获取数据？
【发布时间】：2020-12-27 15:15:07
【问题描述】：

我正在尝试使用 dapper 从 sql server 获取数据。我需要导出存储在 Azure sql 数据库中的 460K 记录。我决定分批获取数据，所以我在每批中记录了 10k 条记录。我计划并行获取记录，因此我将异步方法添加到任务列表中并执行Task.WhenAll。当我在本地运行时代码运行良好，但在部署到 k8s 集群后，我遇到了一些记录的数据读取问题。我是多线程的新手，我不知道如何处理这个问题。我试图在方法内部做一个锁定，但是系统崩溃了，下面是我的代码，代码可能很笨拙，因为我尝试了很多解决方案来解决这个问题。

for (int i = 0; i < numberOfPages; i++)
{
  tableviewWithCondition.startRow = startRow;
  resultData.Add(_tableviewRepository.GetTableviewRowsByPagination(tableviewExportCondition.TableviewName, modelMappingGroups, tableviewWithCondition.startRow, builder, pageSize, appName, i));
  startRow += tableviewWithCondition.pageSize;
}

foreach(var task in resultData)
{ 
   if (task != null)
   {
      dataToExport.AddRange(task.Result);
   }
}

这是我实现的使用 dapper 从 azure sql 数据库获取数据的方法。

public async Task<(IEnumerable<int> unprocessedData, IEnumerable<dynamic> rowData)> GetTableviewRowsByPagination(string tableName, IEnumerable<MappingGroup> tableviewAttributeDetails,
           int startRow, SqlBuilder builder, int pageSize = 100, AppNameEnum appName = AppNameEnum.OptiSoil, int taskNumber = 1)
        {
            var _unitOfWork = _unitOfWorkServices.Build(appName.ToString());
            List<int> unprocessedData = new List<int>();
            try
            {
                var columns = tableviewAttributeDetails.Select(c => { return $"{c.mapping_group_value} [{c.attribute}]"; });
                var joinedColumn = string.Join(",", columns);
                builder.Select(joinedColumn);
                var selector = builder.AddTemplate($"SELECT /**select**/ FROM {tableName} with (nolock) /**innerjoin**/ /**where**/ /**orderby**/ OFFSET {startRow} ROWS FETCH NEXT {(pageSize == 0 ? 100 : pageSize)} ROWS ONLY");

                using (var connection = _unitOfWork.Connection)
                {
                    connection.Open();
                    var data = await connection.QueryAsync(selector.RawSql, selector.Parameters);
                    Console.WriteLine($"data completed for task{taskNumber}");
                    return (unprocessedData, data);
                }
            }
            catch(Exception ex)
            {
                Console.WriteLine($"Exception: {ex.Message}");
                if (ex.InnerException != null)
                    Console.WriteLine($"InnerException: {ex.InnerException.Message}");

                Console.WriteLine($"Error in fetching from row {startRow}");
                unprocessedData.Add(startRow);
                return (unprocessedData, null);
            }
            finally
            {
                _unitOfWork.Dispose();
            }
        }

上面的代码在本地运行良好，但在服务器上我遇到了问题。

异常：向服务器发送请求时发生传输级错误。（提供者：TCP 提供者，错误：35 - 捕获到内部异常）。

InnerException：当另一个写入操作处于挂起状态时，无法调用 WriteAsync 方法。

在并行任务中获取数据时如何避免这个问题？

【问题讨论】：

这些异常在哪里被捕获，它们的堆栈跟踪是什么？
实际上我通过增加 sqlconnection 超时解决了这个问题。我发现在尝试打开连接时会出现此问题。现在它工作正常。

标签： c# multithreading .net-core async-await task-parallel-library

【解决方案1】：

您正在使用相同的连接并尝试在其上执行多个命令（我假设这是因为命名），您还应该处置工作单元吗？

而不是：

 using (var connection = _unitOfWork.Connection)
                {
                    connection.Open();
                    var data = await connection.QueryAsync(selector.RawSql, selector.Parameters);
                    Console.WriteLine($"data completed for task{taskNumber}");
                    return (unprocessedData, data);
                }

为每个项目创建一个新的连接，如果这是您真正想要做的。我想，这是有根据的猜测，由于时间原因，它在本地工作。

还可以查看 Task.WhenAll，这是收集所有结果的更好方法。而不是：

foreach(var task in resultData)
{ 
   if (task != null)
   {
      dataToExport.AddRange(task.Result);
   }
}

在任务上调用结果通常是不好的做法。

【讨论】：