修复mysql数据库MySqlBulkLoader批量插入在linux上数据丢失的问题

cq-panda · cq-panda · commit 0128b78eeb25 · 2021-04-23T00:13:50.000+08:00
diff --git a/Vue.Net/VOL.Core/Dapper/SqlDapper.cs b/Vue.Net/VOL.Core/Dapper/SqlDapper.cs
@@ -6,6 +6,7 @@
 using System.Collections.Generic;
 using System.Data;
 using System.Data.SqlClient;
+using System.IO;
 using System.Linq;
 using System.Linq.Expressions;
 using System.Text;
@@ -110,16 +111,16 @@ public T QueryFirst<T>(string cmd, object param, CommandType? commandType = null
             return QueryList<T>(cmd, param, commandType: commandType ?? CommandType.Text, beginTransaction: beginTransaction).FirstOrDefault();
         }
 
-        public List<dynamic> QueryDynamicList(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false) 
+        public List<dynamic> QueryDynamicList(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false)
         {
             return Execute((conn, dbTransaction) =>
             {
                 return conn.Query<dynamic>(cmd, param, dbTransaction, commandType: commandType ?? CommandType.Text).ToList();
             }, beginTransaction);
         }
-        public dynamic QueryDynamicFirst(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false) 
+        public dynamic QueryDynamicFirst(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false)
         {
-          return QueryList<dynamic>(cmd, param, commandType: commandType ?? CommandType.Text, beginTransaction: beginTransaction).FirstOrDefault();
+            return QueryList<dynamic>(cmd, param, commandType: commandType ?? CommandType.Text, beginTransaction: beginTransaction).FirstOrDefault();
         }
 
         public object ExecuteScalar(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false)
@@ -282,7 +283,7 @@ public int AddRange<T>(IEnumerable<T> entities, Expression<Func<T, object>> addF
             return Execute<int>((conn, dbTransaction) =>
             {
                 //todo pgsql待实现
-                return conn.Execute(sql, (DBType.Name == DbCurrentType.MySql.ToString() || DBType.Name == DbCurrentType.PgSql.ToString()) ? entities.ToList() : null,dbTransaction);
+                return conn.Execute(sql, (DBType.Name == DbCurrentType.MySql.ToString() || DBType.Name == DbCurrentType.PgSql.ToString()) ? entities.ToList() : null, dbTransaction);
             }, beginTransaction);
         }
 
@@ -433,6 +434,7 @@ public int BulkInsert(DataTable table, string tableName, SqlBulkCopyOptions? sql
 
         /// <summary>
         ///大批量数据插入,返回成功插入行数
+        ////************(网上的示例在linux上运行批量插入就是巨坑,会丢数据,至于为什么,见MySqlBulkLoader源码)***************/
         /// </summary>
         /// <param name="connectionString">数据库连接字符串</param>
         /// <param name="table">数据表</param>
@@ -442,11 +444,13 @@ private int MySqlBulkInsert(DataTable table, string tableName, string fileName =
             if (table.Rows.Count == 0)
                 return 0;
             tmpPath = tmpPath ?? FileHelper.GetCurrentDownLoadPath();
-            fileName = fileName ?? $"{DateTime.Now.ToString("yyyyMMddHHmmss")}.csv";
+            //  fileName = fileName ?? $"{DateTime.Now.ToString("yyyyMMddHHmmss")}.csv";
             int insertCount = 0;
             string csv = DataTableToCsv(table);
-            FileHelper.WriteFile(tmpPath, fileName, csv);
-            string path = tmpPath + fileName;
+            // FileHelper.WriteFile(tmpPath, fileName, csv);
+            // string path = tmpPath + fileName;
+            string text = $"当前行:{table.Rows.Count}";
+            MemoryStream stream = null;
             try
             {
                 if (Connection.State == ConnectionState.Closed)
@@ -455,17 +459,21 @@ private int MySqlBulkInsert(DataTable table, string tableName, string fileName =
                 {
                     MySqlBulkLoader bulk = new MySqlBulkLoader(Connection as MySqlConnection)
                     {
-                        FieldTerminator = ",",
-                        FieldQuotationCharacter = '"',
-                        EscapeCharacter = '"',
-                        LineTerminator = "\r\n",
-                        FileName = path.ReplacePath(),
-                        NumberOfLinesToSkip = 0,
+                        LineTerminator = "\n",
                         TableName = tableName,
                         CharacterSet = "UTF8"
                     };
+                    var array = Encoding.UTF8.GetBytes(csv);
+                     stream = new MemoryStream(array);
+                    // StreamReader reader = new StreamReader(stream);
+                    bulk.SourceStream = stream; //File.OpenRead(fileName);
                     bulk.Columns.AddRange(table.Columns.Cast<DataColumn>().Select(colum => colum.ColumnName).ToList());
                     insertCount = bulk.Load();
+                    //text = text + $",实际写入行:{insertCount}";
+                    //fileName = fileName ?? $"{DateTime.Now.ToString("yyyyMMddHHmmss")}.text";
+                    //FileHelper.WriteFile(tmpPath, fileName, text);
+                    //Console.WriteLine(text);
+                    //Console.WriteLine(insertCount);
                     tran.Commit();
                 }
             }
@@ -477,6 +485,7 @@ private int MySqlBulkInsert(DataTable table, string tableName, string fileName =
             {
                 Connection?.Dispose();
                 Connection?.Close();
+                stream?.Dispose();
             }
             return insertCount;
             //   File.Delete(path);
@@ -501,7 +510,7 @@ private string DataTableToCsv(DataTable table)
                 for (int i = 0; i < table.Columns.Count; i++)
                 {
                     colum = table.Columns[i];
-                    if (i != 0) sb.Append(",");
+                    if (i != 0) sb.Append("\t");
                     if (colum.DataType == typeString && row[colum].ToString().Contains(","))
                     {
                         sb.Append("\"" + row[colum].ToString().Replace("\"", "\"\"") + "\"");
@@ -514,7 +523,7 @@ private string DataTableToCsv(DataTable table)
                     }
                     else sb.Append(row[colum].ToString());
                 }
-                sb.AppendLine();
+                sb.Append("\n");
             }
 
             return sb.ToString();
diff --git a/开发版dev/Vue.NetCore/Vue.Net/VOL.Core/Dapper/SqlDapper.cs b/开发版dev/Vue.NetCore/Vue.Net/VOL.Core/Dapper/SqlDapper.cs
@@ -6,6 +6,7 @@
 using System.Collections.Generic;
 using System.Data;
 using System.Data.SqlClient;
+using System.IO;
 using System.Linq;
 using System.Linq.Expressions;
 using System.Text;
@@ -110,16 +111,16 @@ public T QueryFirst<T>(string cmd, object param, CommandType? commandType = null
             return QueryList<T>(cmd, param, commandType: commandType ?? CommandType.Text, beginTransaction: beginTransaction).FirstOrDefault();
         }
 
-        public List<dynamic> QueryDynamicList(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false) 
+        public List<dynamic> QueryDynamicList(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false)
         {
             return Execute((conn, dbTransaction) =>
             {
                 return conn.Query<dynamic>(cmd, param, dbTransaction, commandType: commandType ?? CommandType.Text).ToList();
             }, beginTransaction);
         }
-        public dynamic QueryDynamicFirst(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false) 
+        public dynamic QueryDynamicFirst(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false)
         {
-          return QueryList<dynamic>(cmd, param, commandType: commandType ?? CommandType.Text, beginTransaction: beginTransaction).FirstOrDefault();
+            return QueryList<dynamic>(cmd, param, commandType: commandType ?? CommandType.Text, beginTransaction: beginTransaction).FirstOrDefault();
         }
 
         public object ExecuteScalar(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false)
@@ -282,7 +283,7 @@ public int AddRange<T>(IEnumerable<T> entities, Expression<Func<T, object>> addF
             return Execute<int>((conn, dbTransaction) =>
             {
                 //todo pgsql待实现
-                return conn.Execute(sql, (DBType.Name == DbCurrentType.MySql.ToString() || DBType.Name == DbCurrentType.PgSql.ToString()) ? entities.ToList() : null,dbTransaction);
+                return conn.Execute(sql, (DBType.Name == DbCurrentType.MySql.ToString() || DBType.Name == DbCurrentType.PgSql.ToString()) ? entities.ToList() : null, dbTransaction);
             }, beginTransaction);
         }
 
@@ -433,6 +434,7 @@ public int BulkInsert(DataTable table, string tableName, SqlBulkCopyOptions? sql
 
         /// <summary>
         ///大批量数据插入,返回成功插入行数
+        ////************(网上的示例在linux上运行批量插入就是巨坑,会丢数据,至于为什么,见MySqlBulkLoader源码)***************/
         /// </summary>
         /// <param name="connectionString">数据库连接字符串</param>
         /// <param name="table">数据表</param>
@@ -442,11 +444,13 @@ private int MySqlBulkInsert(DataTable table, string tableName, string fileName =
             if (table.Rows.Count == 0)
                 return 0;
             tmpPath = tmpPath ?? FileHelper.GetCurrentDownLoadPath();
-            fileName = fileName ?? $"{DateTime.Now.ToString("yyyyMMddHHmmss")}.csv";
+            //  fileName = fileName ?? $"{DateTime.Now.ToString("yyyyMMddHHmmss")}.csv";
             int insertCount = 0;
             string csv = DataTableToCsv(table);
-            FileHelper.WriteFile(tmpPath, fileName, csv);
-            string path = tmpPath + fileName;
+            // FileHelper.WriteFile(tmpPath, fileName, csv);
+            // string path = tmpPath + fileName;
+            string text = $"当前行:{table.Rows.Count}";
+            MemoryStream stream = null;
             try
             {
                 if (Connection.State == ConnectionState.Closed)
@@ -455,17 +459,21 @@ private int MySqlBulkInsert(DataTable table, string tableName, string fileName =
                 {
                     MySqlBulkLoader bulk = new MySqlBulkLoader(Connection as MySqlConnection)
                     {
-                        FieldTerminator = ",",
-                        FieldQuotationCharacter = '"',
-                        EscapeCharacter = '"',
-                        LineTerminator = "\r\n",
-                        FileName = path.ReplacePath(),
-                        NumberOfLinesToSkip = 0,
+                        LineTerminator = "\n",
                         TableName = tableName,
                         CharacterSet = "UTF8"
                     };
+                    var array = Encoding.UTF8.GetBytes(csv);
+                     stream = new MemoryStream(array);
+                    // StreamReader reader = new StreamReader(stream);
+                    bulk.SourceStream = stream; //File.OpenRead(fileName);
                     bulk.Columns.AddRange(table.Columns.Cast<DataColumn>().Select(colum => colum.ColumnName).ToList());
                     insertCount = bulk.Load();
+                    //text = text + $",实际写入行:{insertCount}";
+                    //fileName = fileName ?? $"{DateTime.Now.ToString("yyyyMMddHHmmss")}.text";
+                    //FileHelper.WriteFile(tmpPath, fileName, text);
+                    //Console.WriteLine(text);
+                    //Console.WriteLine(insertCount);
                     tran.Commit();
                 }
             }
@@ -477,6 +485,7 @@ private int MySqlBulkInsert(DataTable table, string tableName, string fileName =
             {
                 Connection?.Dispose();
                 Connection?.Close();
+                stream?.Dispose();
             }
             return insertCount;
             //   File.Delete(path);
@@ -501,7 +510,7 @@ private string DataTableToCsv(DataTable table)
                 for (int i = 0; i < table.Columns.Count; i++)
                 {
                     colum = table.Columns[i];
-                    if (i != 0) sb.Append(",");
+                    if (i != 0) sb.Append("\t");
                     if (colum.DataType == typeString && row[colum].ToString().Contains(","))
                     {
                         sb.Append("\"" + row[colum].ToString().Replace("\"", "\"\"") + "\"");
@@ -514,7 +523,7 @@ private string DataTableToCsv(DataTable table)
                     }
                     else sb.Append(row[colum].ToString());
                 }
-                sb.AppendLine();
+                sb.Append("\n");
             }
 
             return sb.ToString();

Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,7 @@`
`6`	`6`	`using System.Collections.Generic;`
`7`	`7`	`using System.Data;`
`8`	`8`	`using System.Data.SqlClient;`
	`9`	`+using System.IO;`
`9`	`10`	`using System.Linq;`
`10`	`11`	`using System.Linq.Expressions;`
`11`	`12`	`using System.Text;`
`@@ -110,16 +111,16 @@ public T QueryFirst<T>(string cmd, object param, CommandType? commandType = null`
`110`	`111`	`return QueryList<T>(cmd, param, commandType: commandType ?? CommandType.Text, beginTransaction: beginTransaction).FirstOrDefault();`
`111`	`112`	`}`
`112`	`113`
`113`		`- public List<dynamic> QueryDynamicList(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false)`
	`114`	`+ public List<dynamic> QueryDynamicList(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false)`
`114`	`115`	`{`
`115`	`116`	`return Execute((conn, dbTransaction) =>`
`116`	`117`	`{`
`117`	`118`	`return conn.Query<dynamic>(cmd, param, dbTransaction, commandType: commandType ?? CommandType.Text).ToList();`
`118`	`119`	`}, beginTransaction);`
`119`	`120`	`}`
`120`		`- public dynamic QueryDynamicFirst(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false)`
	`121`	`+ public dynamic QueryDynamicFirst(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false)`
`121`	`122`	`{`
`122`		`- return QueryList<dynamic>(cmd, param, commandType: commandType ?? CommandType.Text, beginTransaction: beginTransaction).FirstOrDefault();`
	`123`	`+ return QueryList<dynamic>(cmd, param, commandType: commandType ?? CommandType.Text, beginTransaction: beginTransaction).FirstOrDefault();`
`123`	`124`	`}`
`124`	`125`
`125`	`126`	`public object ExecuteScalar(string cmd, object param, CommandType? commandType = null, bool beginTransaction = false)`
`@@ -282,7 +283,7 @@ public int AddRange<T>(IEnumerable<T> entities, Expression<Func<T, object>> addF`
`282`	`283`	`return Execute<int>((conn, dbTransaction) =>`
`283`	`284`	`{`
`284`	`285`	`//todo pgsql待实现`
`285`		`- return conn.Execute(sql, (DBType.Name == DbCurrentType.MySql.ToString() \|\| DBType.Name == DbCurrentType.PgSql.ToString()) ? entities.ToList() : null,dbTransaction);`
	`286`	`+ return conn.Execute(sql, (DBType.Name == DbCurrentType.MySql.ToString() \|\| DBType.Name == DbCurrentType.PgSql.ToString()) ? entities.ToList() : null, dbTransaction);`
`286`	`287`	`}, beginTransaction);`
`287`	`288`	`}`
`288`	`289`
`@@ -433,6 +434,7 @@ public int BulkInsert(DataTable table, string tableName, SqlBulkCopyOptions? sql`
`433`	`434`
`434`	`435`	`/// <summary>`
`435`	`436`	`///大批量数据插入,返回成功插入行数`
	`437`	`+ ////**********(网上的示例在linux上运行批量插入就是巨坑,会丢数据,至于为什么,见MySqlBulkLoader源码)*************/`
`436`	`438`	`/// </summary>`
`437`	`439`	`/// <param name="connectionString">数据库连接字符串</param>`
`438`	`440`	`/// <param name="table">数据表</param>`
`@@ -442,11 +444,13 @@ private int MySqlBulkInsert(DataTable table, string tableName, string fileName =`
`442`	`444`	`if (table.Rows.Count == 0)`
`443`	`445`	`return 0;`
`444`	`446`	`tmpPath = tmpPath ?? FileHelper.GetCurrentDownLoadPath();`
`445`		`- fileName = fileName ?? $"{DateTime.Now.ToString("yyyyMMddHHmmss")}.csv";`
	`447`	`+ // fileName = fileName ?? $"{DateTime.Now.ToString("yyyyMMddHHmmss")}.csv";`
`446`	`448`	`int insertCount = 0;`
`447`	`449`	`string csv = DataTableToCsv(table);`
`448`		`- FileHelper.WriteFile(tmpPath, fileName, csv);`
`449`		`- string path = tmpPath + fileName;`
	`450`	`+ // FileHelper.WriteFile(tmpPath, fileName, csv);`
	`451`	`+ // string path = tmpPath + fileName;`
	`452`	`+ string text = $"当前行:{table.Rows.Count}";`
	`453`	`+ MemoryStream stream = null;`
`450`	`454`	`try`
`451`	`455`	`{`
`452`	`456`	`if (Connection.State == ConnectionState.Closed)`
`@@ -455,17 +459,21 @@ private int MySqlBulkInsert(DataTable table, string tableName, string fileName =`
`455`	`459`	`{`
`456`	`460`	`MySqlBulkLoader bulk = new MySqlBulkLoader(Connection as MySqlConnection)`
`457`	`461`	`{`
`458`		`- FieldTerminator = ",",`
`459`		`- FieldQuotationCharacter = '"',`
`460`		`- EscapeCharacter = '"',`
`461`		`- LineTerminator = "\r\n",`
`462`		`- FileName = path.ReplacePath(),`
`463`		`- NumberOfLinesToSkip = 0,`
	`462`	`+ LineTerminator = "\n",`
`464`	`463`	`TableName = tableName,`
`465`	`464`	`CharacterSet = "UTF8"`
`466`	`465`	`};`
	`466`	`+ var array = Encoding.UTF8.GetBytes(csv);`
	`467`	`+ stream = new MemoryStream(array);`
	`468`	`+ // StreamReader reader = new StreamReader(stream);`
	`469`	`+ bulk.SourceStream = stream; //File.OpenRead(fileName);`
`467`	`470`	`bulk.Columns.AddRange(table.Columns.Cast<DataColumn>().Select(colum => colum.ColumnName).ToList());`
`468`	`471`	`insertCount = bulk.Load();`
	`472`	`+ //text = text + $",实际写入行:{insertCount}";`
	`473`	`+ //fileName = fileName ?? $"{DateTime.Now.ToString("yyyyMMddHHmmss")}.text";`
	`474`	`+ //FileHelper.WriteFile(tmpPath, fileName, text);`
	`475`	`+ //Console.WriteLine(text);`
	`476`	`+ //Console.WriteLine(insertCount);`
`469`	`477`	`tran.Commit();`
`470`	`478`	`}`
`471`	`479`	`}`
`@@ -477,6 +485,7 @@ private int MySqlBulkInsert(DataTable table, string tableName, string fileName =`
`477`	`485`	`{`
`478`	`486`	`Connection?.Dispose();`
`479`	`487`	`Connection?.Close();`
	`488`	`+ stream?.Dispose();`
`480`	`489`	`}`
`481`	`490`	`return insertCount;`
`482`	`491`	`// File.Delete(path);`
`@@ -501,7 +510,7 @@ private string DataTableToCsv(DataTable table)`
`501`	`510`	`for (int i = 0; i < table.Columns.Count; i++)`
`502`	`511`	`{`
`503`	`512`	`colum = table.Columns[i];`
`504`		`- if (i != 0) sb.Append(",");`
	`513`	`+ if (i != 0) sb.Append("\t");`
`505`	`514`	`if (colum.DataType == typeString && row[colum].ToString().Contains(","))`
`506`	`515`	`{`
`507`	`516`	`sb.Append("\"" + row[colum].ToString().Replace("\"", "\"\"") + "\"");`
`@@ -514,7 +523,7 @@ private string DataTableToCsv(DataTable table)`
`514`	`523`	`}`
`515`	`524`	`else sb.Append(row[colum].ToString());`
`516`	`525`	`}`
`517`		`- sb.AppendLine();`
	`526`	`+ sb.Append("\n");`
`518`	`527`	`}`
`519`	`528`
`520`	`529`	`return sb.ToString();`