背景
在介绍整个原理和代码之前，我们来交代一下整个过程的背景，我们的系统作为一个下游的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:testpassword@192.168.17.42:21/CheryDcsParts",
"BranchPartStockInfo": "ftp://testuser:testpassword@192.168.17.42: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一次插入大量数据到数据库中。