Asp.Net Core中使用FTP读取大文件并使用SqlBulkCopy实现大批量插入SQL SERVER数据库

背景

在介绍整个原理和代码之前，我们来交代一下整个过程的背景，我们的系统作为一个下游的DMS系统，需要每天定期读取第三方接口传入数据，由于第三方接口每天传入的数据有上百万条主机厂备件库存数据，之前通过Kafka消息接口进行传输的话效率低、速度慢而且容易出错，所以后面和第三方统一意见采用FTP方式进行发送和接收，这样我们只需每天去他们的FTP服务器上面去读取他们的文件并实时插入到我们的SQL SERVER数据库中即可，整个过程的重点就是两个部分：1 从第三方FTP服务器读取大文件 2 大量数据插入SQL SERVER数据库，后面的部分重点也是这两个部分，一是要能够快速读取这些数据，二是要能够快速插入数据库而不能够拖慢数据库性能。

一 FTP服务器读取大文件

　　1.1 安装Nuget包

第一个部分就是怎样从第三方的FTP服务器上面读取到对应的文件信息呢？Asp.Net Core中有哪些比较成熟的方案呢？通过调研我们发现FluentFTP这个包能够完全适应我们的要求，所以第一步就是项目中安装FluentFTP这个包，具体的一些介绍也可以参考README.md上面的介绍，安装了这个包之后就是具体的配置和代码读取文件的方式了。

1.2 配置FTP相关地址

和所有其它远程登录形式相同，这个包也是通过FTPClient的形式实现的，所以第一步我们就需要配置具体的FTP服务的地址，这个在appsettings.json中进行配置(这个在实际的生产环境也可以配置在响应的k8s的pod上面)，这个配置包括服务器地址、用户名和密码等相关信息，如下面所示：

"Eai": {
    "Addresses": {
      "PartInfo": "ftp://testuser::21/CheryDcsParts",
      "BranchPartStockInfo": "ftp://testuser::21/CheryDcsParts"
    }
  }

1.3 读取FTP服务器上面的文件

这个部分我们主要是通过具体的代码来说明，这里特别需要说明，下面的代码中我们会将FTP文件读取到内容并插入到数据库中，这个子函数的参数由两部分构成，filePath和instanceId两个，这个是有第三方接口主动调用我们的WebApi接口时传输的，由于读取的文件可能存在各种错误，所以我们进行一行行读取的时候会判断当前行是否有错误，我们会将整个结果分成两个部分，一个是正确的数据，另外一个部分是错误的数据，这两个部分最终会插入到数据库中不同的表中进行保存，便于日后排查数据。

public async Task<FtpResultModel> GetResultFromFtpAsync(string filePath, string instanceId) {
            var inputs = new List<BranchPartStockTempModel>();
            var storagePath = _configuration.GetValue<string>("Local:BranchPartStockInfo");
            var ftpBasePath = new UriBuilder(_configuration.GetValue<string>("EAI:Addresses:BranchPartStockInfo"));
            Encoding.RegisterProvider(CodePagesEncodingProvider.Instance);
            //可以对中文进行编码
            var gb2312Encoding = Encoding.GetEncoding("GB2312");
            // 受限于 Kubernetes Network，无法使用 Active Mode 访问 FTP 服务端
            // 完整的文件路径 = FTP 用户的当前目录 + filePath
            IFtpClient ftpClient = new FtpClient(ftpBasePath.Host, ftpBasePath.Port, ftpBasePath.UserName, ftpBasePath.Password) {
                DataConnectionType = FtpDataConnectionType.PASV
            };

            //使用浏览器，查看ftp路径为ftp://10.2.14.237/interface/erp/zsdeai2016/
            //通过GetWorkingDirectory(),会获取到 /oracle/eaiftp/ 导致路径错误
            //var remoteFilePath = Path.Combine(ftpClient.GetWorkingDirectory(), filePath);
            var remoteFilePath = Path.Combine(ftpBasePath.Path, filePath);
            _logger.LogInformation("Loading file {0} from FTP.", remoteFilePath);
            using var fileStream = await ftpClient.OpenReadAsync(remoteFilePath);
            using var reader = new StreamReader(fileStream, gb2312Encoding);
            //备份文件
            var localFilePath = CombinePath(storagePath, $"instanceId.{Path.GetRandomFileName()}.txt");
            //判断目录是否存在
            if (!Directory.Exists(storagePath)) {
                Directory.CreateDirectory(storagePath);
            }

            using var fileStreamBack = new StreamWriter(localFilePath);
            //错误数据日志
            var errorLogs = new List<BranchPartStockSyncErrorLog>();
            //解析数据
            string line;
            var i = 0;
            while ((line = await reader.ReadLineAsync()) != null) {
                var hasError = false;
                await fileStreamBack.WriteLineAsync(line);
                ++i;
                if (string.IsNullOrWhiteSpace(line))
                    continue;
                var value = line.Split(‘$‘, StringSplitOptions.RemoveEmptyEntries);

                if (value.Length < 4) {
                    CreatePartSyncLog(filePath, instanceId, errorLogs, i, null, null, SharedLocalizer["当前行，数据结构有误。缺少列"]);
                } else {
                    //数据结构：第一列是件号，第二列是公司（集团），第三列是仓库，第四列是库存数量
                    var partCode = GetValueByIndex(value, 0);
                    var warehouseCode = GetValueByIndex(value, 2);

                    if (string.IsNullOrWhiteSpace(warehouseCode)) {
                        hasError = true;
                        CreatePartSyncLog(filePath, instanceId, errorLogs, i, warehouseCode, partCode, SharedLocalizer["仓库编号不能为空"]);
                    }

                    if (string.IsNullOrWhiteSpace(partCode)) {
                        hasError = true;
                        CreatePartSyncLog(filePath, instanceId, errorLogs, i, warehouseCode, partCode, SharedLocalizer["配件名称不能为空"]);
                    }

                    //获取的库存数量
                    var stockQuantityStr = GetValueByIndex(value, 3);

                    if (!decimal.TryParse(stockQuantityStr, out var stockQuantity)) {
                        hasError = true;
                        CreatePartSyncLog(filePath, instanceId, errorLogs, i, warehouseCode, partCode, SharedLocalizer["传入的库存数量格式不正确"]);
                    }

                    if (hasError) {
                        continue;
                    }

                    var branchPartStockTempModel = new BranchPartStockTempModel {
                        WarehouseCode = warehouseCode,
                        PartCode = partCode,
                        StockQuantity = stockQuantity,
                        LineNumber = i
                    };

                    inputs.Add(branchPartStockTempModel);
                }
            }

            _logger.LogInformation($"收到Sap备件, 数量 {inputs.Count + errorLogs.Count},其中正确：{inputs.Count}条，错误：{errorLogs.Count}条");

            return new FtpResultModel {
                ValidData = inputs,
                ErrorLogs = errorLogs
            };
        }

这个里面还有一点需要注意的地方就是，我们读取的FileStream以及StreamReader都是.Net 中非托管的对象，需要在使用之后进行手动释放，否则会造成无法及时释放内存造成内存使用率增加，这个在使用的时候需要特别注意，所以在我们的代码中都使用了using前缀从而在使用完毕后及时释放非托管内存。此外我们的代码中还是需要使用ILogger及时记录一些关键信息从而方便线上环境进行问题排查。

　　二 使用SqlBulkCopy实现大批量插入SQL SERVER数据库

在上面的部分我们已经将FTP服务器上面的数据读取到了内存中，接下来我们就演示如何将我们特定的数据通过SqlBulkCopy插入到数据库中,在开始之前你也可以点击这里来阅读SqlBulkCopy的官方文档。

这个部分我们也是通过下面的代码来一步步讲述。

/// <summary>
        /// 插入分公司备件库存
        /// </summary>
        /// <param name="filePath">文件路径</param>
        /// <param name="instanceId">实例Id</param>
        /// <param name="branchPartStockModels">分公司备件库存实体</param>
        /// <param name="errorBranchPartStockModels">错误信息实体</param>
        /// <returns></returns>
        [UnitOfWork(isTransactional: false)]
        public async Task InsertBranchPartStockAsync(string filePath, string instanceId, IList<SyncGetBranchPartStockModel> branchPartStockModels, IList<BranchPartStockSyncErrorLog> errorBranchPartStockModels) {
            var transaction = Connection.BeginTransaction();
            var command = Connection.CreateCommand();
            try {
                const int CommandTimeOut = 5 * 60;
                command.Transaction = transaction;
                command.CommandText = _tableGenerator.GenerateTableScript(typeof(SyncGetBranchPartStockModel), DataBaseType.SqlServer);
                command.CommandTimeout = CommandTimeOut;
                await command.ExecuteNonQueryAsync();
                await new BulkUploadToSql().CommitAsync(branchPartStockModels, transaction);
                command.CommandText = MergeBranchPartStock(filePath, instanceId);
                await command.ExecuteNonQueryAsync();

                await new BulkUploadToSql().CommitAsync(errorBranchPartStockModels, transaction);

                transaction.Commit();
            } finally {
                command?.Dispose();
                transaction?.Dispose();
            }
        }

这段代码中Connection是基类中对外公开的一个属性，指的是当前DbContext对应的唯一的Connection,这里面所有的操作都是通过ExecuteNonQuery这个来实现的这个大家应该非常熟悉了，这段代码中还使用到了_tableGenerator.GenerateTableScript这个方法，这个方法主要是通过定义一个实体Model用代码来创建数据库临时表方法，如果想了解更多内容，请点击这里查看，这个里面最核心的部分就是下面的这段代码了。

new BulkUploadToSql().CommitAsync(branchPartStockModels, transaction);

这里我们来重点看看我们在BulkUploadToSql的方法中到底实现了些什么？

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.ComponentModel.DataAnnotations;
using System.ComponentModel.DataAnnotations.Schema;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.Reflection;
using System.Threading.Tasks;
using Sunlight.DataAnnotations;

namespace Sunlight.Dcs.Parts.Data.EntityFrameworkCore {
    public class BulkUploadToSql : IBulkUploadToSql {
        public string TableName { get; set; }
        public int CommitBatchSize { get; set; } = 1000;

        public async Task CommitAsync<T>(IList<T> sourceData, IDbTransaction transaction) {
            if (sourceData.Count <= 0) {
                return;
            }
            GetTableName(typeof(T));

            // make sure to enable triggers
            // more on triggers in next post
            if (!(transaction.Connection is SqlConnection connection))
                throw new ValidationException("当前只支持SqlServer, 其它数据库需要开发");

            if (connection.State != ConnectionState.Open)
                connection.Open();
            using var bulkCopy = new SqlBulkCopy(connection,
                SqlBulkCopyOptions.CheckConstraints | SqlBulkCopyOptions.KeepNulls, (SqlTransaction)transaction
            ) { DestinationTableName = TableName };

            using var dt = new DataTable();
            var numberOfPages = (sourceData.Count / CommitBatchSize) + (sourceData.Count % CommitBatchSize == 0 ? 0 : 1);
            for (var pageIndex = 0; pageIndex < numberOfPages; pageIndex++) {
                sourceData.Skip(pageIndex * CommitBatchSize).Take(CommitBatchSize).ToDataTable(dt);
                // write the data in the "dataTable"
                await bulkCopy.WriteToServerAsync(dt);
                // reset
                //this.dataTable.Clear();
            }
        }

        /// <summary>
        ///
        /// </summary>
        /// <returns></returns>
        private void GetTableName(MemberInfo entityType) {
            if (TableName != null)
                return;
            var prefix = entityType.GetCustomAttribute<TempTableAttribute>() != null ? "#" : string.Empty;
            TableName = $"{prefix}{entityType.GetCustomAttribute<TableAttribute>()?.Name ?? entityType.Name}";
        }
    }

    public static class BulkUploadToSqlHelper {
        public static void ToDataTable<T>(this IEnumerable<T> data, DataTable table) {
            table.Reset();
            var properties =
                TypeDescriptor.GetProperties(typeof(T));
            foreach (PropertyDescriptor prop in properties)
                table.Columns.Add(prop.Name, Nullable.GetUnderlyingType(prop.PropertyType) ?? prop.PropertyType);
            foreach (var item in data) {
                var row = table.NewRow();
                foreach (PropertyDescriptor prop in properties)
                    row[prop.Name] = prop.GetValue(item) ?? DBNull.Value;
                table.Rows.Add(row);
            }
        }
    }
}

这个里面CommitAsync是整个类核心，这个方法里面的sourceData就是我们从FTP服务器上面读取到的大批量文件，大概有100万条数据，这么多的数据如果通过Insert进行一条一条的插入那简直就是一种灾难。在我们的代码中我们也是通过创建一个SqlBulkCopy对象来分页插入的，每次都是插入1000条，当然每次插入的数量也是可以根据我们的需要来进行调整的，这个里面我们可以看一下微软官方文档中最重要的一段对SqlBulkCopy中的描述：

SqlBulkCopy 类可用于只将数据写入 SQL Server 表。 但是，数据源不限于 SQL Server;可以使用任何数据源，只要数据可以加载到 DataTable 实例或使用 IDataReader 实例进行读取即可。

　　所以我们的第一步工作就是将我们内存中的数据读取到DataTable中去，后面我们就能够将DataTable中的数据实现批量拷贝到数据库中的目的了。

通过上面的两个过程我们就能够完整地表现这篇博客的主题，主要包括：1 如何使用FluentFTP来读取文件。2 如何使用SqlBulkCopy一次插入大量数据到数据库中。