OracleBulkCopy
Posted 听哥哥的话
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了OracleBulkCopy相关的知识，希望对你有一定的参考价值。
using Framework; using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using System.Data; using NHibernate; using Oracle.DataAccess.Client; using System.IO; using System.Diagnostics; namespace Holworth.Utility.BulkCopy { public class OracleBulkCopy : BulkCopyTemplate, IBulkCopy { private static object lockBulkCopy = new object(); private static IBulkCopy instance; public static IBulkCopy Instance { get { if (instance == null) { lock (lockBulkCopy) { instance = new OracleBulkCopy(); } } return instance; } } private Spring.Context.IApplicationContext _ctx; public Spring.Context.IApplicationContext ctx { get { if (_ctx == null) { _ctx = Spring.Context.Support.ContextRegistry.GetContext(); } return _ctx; } } private Framework.IService.ICommonService _dao; public Framework.IService.ICommonService Dao { get { if (_dao == null) { _dao = (Framework.IService.ICommonService)ctx["CommonService"]; } return _dao; } set { _dao = value; } } private static object lockSequence = new object(); public override long GetSequenceCurrentVal(string v_seq) { long inCreId = 1; try { QueryInfo searchInfo = new QueryInfo(); searchInfo.CustomSQL = $"select {v_seq}.NEXTVAL from dual"; var table = Dao.ExecuteDataSet(searchInfo).Tables[0]; inCreId = Convert.ToInt64(table.Rows[0][0].ToString()); } catch (Exception ex) { try { var info = new QueryInfo(); info.CustomSQL = $"create sequence {v_seq} minvalue 1 maxvalue 999999999999999 start with 1 increment by 1 cache 20"; Dao.ExecuteNonQuery(info); var searchInfo = new QueryInfo(); searchInfo.CustomSQL = $"select {v_seq}.NEXTVAL from dual"; var table = Dao.ExecuteDataSet(searchInfo).Tables[0]; inCreId = Convert.ToInt64(table.Rows[0][0].ToString()); } catch (Exception e) { Logger.Fatal(e); throw e; } } return inCreId; } public override void ChangeSequenceStep(string v_seq, int rowCount) { var info = new QueryInfo(); info.NamedQuery = "PRO_SEQUENCE"; info.Parameters.Add("v_simulation_number", rowCount); info.Parameters.Add("v_seq", v_seq); Dao.ExecuteNonQuery(info); } public override void SequenceStepReset(string v_seq) { var info = new QueryInfo(); info.NamedQuery = "PRO_SEQUENCE"; info.Parameters.Add("v_simulation_number", -1); info.Parameters.Add("v_seq", v_seq); Dao.ExecuteNonQuery(info); } public override void BulkCopyImport(object bulkcopy, string targetTable, string IdField, DataTable dt, BulkCopyParam bp = default(BulkCopyParam)) { try { if (dt.Rows.Count > 0) { var session = Dao.GetCurrentSession(); object cname = ""; var sql = ""; #region oracle 查看主键约束的名称 sql = $@"select a.constraint_name c,a.column_name from user_cons_columns a,user_constraints b where a.constraint_name=b.constraint_name and b.constraint_type=‘P‘ and b.owner=‘{Holworth.Utility.HraUtility.GetDataBaseInfo(AppEnum.Web).user}‘ and a.table_name=‘{targetTable.ToUpper()}‘"; #endregion try { cname = session.CreateSQLQuery(sql).AddScalar("c", NHibernateUtil.String).UniqueResult(); if (cname != null) { sql = $@"alter table {targetTable} drop constraint {cname}"; } if (bp != null && bp.Generator != GenerateValue.UUID) session.CreateSQLQuery(sql).ExecuteUpdate(); } catch (Exception ex) { HttpMessages.Instance().PushMessage(ex.Message + "详细" + ex.StackTrace); Logger.Fatal(ex); } BulkCopyImport2(targetTable, IdField, dt, bp); try { if (cname != null) { sql = $@"alter table {targetTable} add constraint {cname} primary key({IdField})"; } else { int len = targetTable.Length > 27 ? 27 : targetTable.Length; sql = $@"alter table {targetTable} add constraint pk_{targetTable.Substring(0, len)} primary key({IdField})"; } if (bp != null && bp.Generator != GenerateValue.UUID) session.CreateSQLQuery(sql).ExecuteUpdate(); } catch (Exception ex) { Logger.Fatal(ex); } } } catch (Exception ex) { Logger.Fatal(ex); HttpMessages.Instance().PushMessage("BULKCOPY表:" + targetTable + "异常因为：" + ex.Message + "堆栈信息:" + ex.StackTrace); throw ex; } } //public override void BulkCopyImport(object bulkcopy, string targetTable, string IdField, DataTable dt, BulkCopyParam bp = default(BulkCopyParam)) //{ // try // { // Oracle.DataAccess.Client.OracleBulkCopy copy = bulkcopy as Oracle.DataAccess.Client.OracleBulkCopy; // if (dt != null && dt.Rows.Count > 0) // { // copy.DestinationTableName = targetTable; // copy.BulkCopyTimeout = 500; // copy.BatchSize = 1000; // } // if (dt.Rows.Count > 0) // { // var session = Dao.GetCurrentSession(); // object cname = ""; // var sql = ""; // #region oracle 查看主键约束的名称 // sql = [email protected]"select a.constraint_name c,a.column_name // from user_cons_columns a,user_constraints b // where a.constraint_name=b.constraint_name // and b.constraint_type=‘P‘ // and b.owner=‘{Holworth.Utility.HraUtility.GetDataBaseInfo(AppEnum.Web).user}‘ // and a.table_name=‘{targetTable.ToUpper()}‘"; // #endregion // try // { // cname = session.CreateSQLQuery(sql).AddScalar("c", NHibernateUtil.String).UniqueResult(); // if (cname != null) // { // sql = [email protected]"alter table {targetTable} drop constraint {cname}"; // } // if (bp != null && bp.Generator != GenerateValue.UUID) // session.CreateSQLQuery(sql).ExecuteUpdate(); // } // catch (Exception ex) // { // HttpMessages.Instance().PushMessage(ex.Message + "详细" + ex.StackTrace); // Logger.Fatal(ex); // } // copy.WriteToServer(dt); // try // { // if (cname != null) // { // sql = [email protected]"alter table {targetTable} add constraint {cname} primary key({IdField})"; // } // else // { // int len = targetTable.Length > 27 ? 27 : targetTable.Length; // sql = [email protected]"alter table {targetTable} add constraint pk_{targetTable.Substring(0, len)} primary key({IdField})"; // } // if (bp != null && bp.Generator != GenerateValue.UUID) // session.CreateSQLQuery(sql).ExecuteUpdate(); // } // catch (Exception ex) // { // Logger.Fatal(ex); // } // } // } // catch (Exception ex) // { // Logger.Fatal(ex); // HttpMessages.Instance().PushMessage("BULKCOPY表:" + targetTable + "异常因为：" + ex.Message + "堆栈信息:" + ex.StackTrace); // throw ex; // } //} public void BulkCopyImport2(string targetTable, string IdField, DataTable dt, BulkCopyParam bp = default(BulkCopyParam)) { var adotemplate = Holworth.Utility.HraUtility.GetAdoTemplate(); //设置一个数据库的连接串 int recc = dt.Rows.Count; string connectStr = adotemplate.DbProvider.ConnectionString; //"User ID=RAMS_XY_TO_GARY;Password=RAMS_XY_TO_GARY;Data Source=(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=192.168.100.100)(PORT=1521)))(CONNECT_DATA=(SERVER=DEDICATED)(SID=HRA)))"; using (OracleConnection conn = new OracleConnection(connectStr)) { try { OracleCommand command = new OracleCommand(); command.Connection = conn; //到此为止，还都是我们熟悉的代码，下面就要开始喽 //这个参数需要指定每次批插入的记录数 command.ArrayBindCount = recc; command.BindByName = true; command.CommandType = CommandType.Text; command.CommandTimeout = 600; //在这个命令行中,用到了参数,参数我们很熟悉,但是这个参数在传值的时候 //用到的是数组,而不是单个的值,这就是它独特的地方 string columns = ""; OracleDbType type = 0; Dictionary<string, OracleDbType> CsharpTypeMappingOracleDbTypeDict = new Dictionary<string, OracleDbType>(); var columnsInfo = Holworth.Utility.ETLColumnInfoDaoController.Current.GetAllTableColumnsInfo().Where(x => string.Compare(targetTable, x.TableName, true) == 0).ToList(); Dictionary<string, List<object>> valueDict = new Dictionary<string, List<object>>(); var paramDict = new Dictionary<string, OracleParameter>(); foreach (ETLColumnTypeInfo ect in columnsInfo) { OracleDbType oracleType = OracleDbType.Varchar2; if (string.Compare(ect.SqlType, "number", true) == 0) { if (ect.Scale > 0) { oracleType = OracleDbType.Decimal; } else { oracleType = OracleDbType.Int32; } } else if (string.Compare(ect.SqlType, OracleDbType.Blob.ToString(), true) == 0) { oracleType = OracleDbType.Blob; } else if (string.Compare(ect.SqlType, OracleDbType.Byte.ToString(), true) == 0) { oracleType = OracleDbType.Byte; } else if (string.Compare(ect.SqlType, OracleDbType.Char.ToString(), true) == 0) { oracleType = OracleDbType.Char; } else if (string.Compare(ect.SqlType, OracleDbType.Clob.ToString(), true) == 0) { oracleType = OracleDbType.Clob; } else if (string.Compare(ect.SqlType, OracleDbType.Date.ToString(), true) == 0) { oracleType = OracleDbType.Date; } else if (string.Compare(ect.SqlType, OracleDbType.Decimal.ToString(), true) == 0) { oracleType = OracleDbType.Decimal; } else if (string.Compare(ect.SqlType, OracleDbType.Double.ToString(), true) == 0) { oracleType = OracleDbType.Double; } else if (string.Compare(ect.SqlType, OracleDbType.Long.ToString(), true) == 0) { oracleType = OracleDbType.Long; } else if (string.Compare(ect.SqlType, OracleDbType.LongRaw.ToString(), true) == 0) { oracleType = OracleDbType.LongRaw; } else if (string.Compare(ect.SqlType, OracleDbType.Int16.ToString(), true) == 0) { oracleType = OracleDbType.Int16; } else if (string.Compare(ect.SqlType, OracleDbType.Int32.ToString(), true) == 0) { oracleType = OracleDbType.Int32; } else if (string.Compare(ect.SqlType, OracleDbType.Int64.ToString(), true) == 0) { oracleType = OracleDbType.Int64; } else if (string.Compare(ect.SqlType, OracleDbType.IntervalDS.ToString(), true) == 0) { oracleType = OracleDbType.IntervalDS; } else if (string.Compare(ect.SqlType, OracleDbType.IntervalYM.ToString(), true) == 0) { oracleType = OracleDbType.IntervalYM; } else if (string.Compare(ect.SqlType, OracleDbType.NClob.ToString(), true) == 0) { oracleType = OracleDbType.NClob; } else if (string.Compare(ect.SqlType, OracleDbType.NChar.ToString(), true) == 0) { oracleType = OracleDbType.NChar; } else if (string.Compare(ect.SqlType, OracleDbType.NVarchar2.ToString(), true) == 0) { oracleType = OracleDbType.NVarchar2; } else if (string.Compare(ect.SqlType, OracleDbType.Raw.ToString(), true) == 0) { oracleType = OracleDbType.Raw; } else if (string.Compare(ect.SqlType, OracleDbType.RefCursor.ToString(), true) == 0) { oracleType = OracleDbType.RefCursor; } else if (string.Compare(ect.SqlType, OracleDbType.Single.ToString(), true) == 0) { oracleType = OracleDbType.Single; } else if (string.Compare(ect.SqlType, OracleDbType.TimeStamp.ToString(), true) == 0) { oracleType = OracleDbType.TimeStamp; } else if (string.Compare(ect.SqlType, OracleDbType.TimeStampLTZ.ToString(), true) == 0) { oracleType = OracleDbType.TimeStampLTZ; } else if (string.Compare(ect.SqlType, OracleDbType.TimeStampTZ.ToString(), true) == 0) { oracleType = OracleDbType.TimeStampTZ; } else if (string.Compare(ect.SqlType, OracleDbType.Varchar2.ToString(), true) == 0) { oracleType = OracleDbType.Varchar2; } else if (string.Compare(ect.SqlType, OracleDbType.XmlType.ToString(), true) == 0) { oracleType = OracleDbType.XmlType; } else if (string.Compare(ect.SqlType, OracleDbType.Array.ToString(), true) == 0) { oracleType = OracleDbType.Array; } else if (string.Compare(ect.SqlType, OracleDbType.Object.ToString(), true) == 0) { oracleType = OracleDbType.Object; } else if (string.Compare(ect.SqlType, OracleDbType.Ref.ToString(), true) == 0) { oracleType = OracleDbType.Ref; } else if (string.Compare(ect.SqlType, OracleDbType.BinaryDouble.ToString(), true) == 0) { oracleType = OracleDbType.BinaryDouble; } else if (string.Compare(ect.SqlType, OracleDbType.BinaryFloat.ToString(), true) == 0) { oracleType = OracleDbType.BinaryFloat; } else { } OracleParameter param = new OracleParameter(ect.ColumnName, oracleType); param.Direction = ParameterDirection.Input; command.Parameters.Add(param); paramDict[ect.ColumnName] = param; valueDict[ect.ColumnName] = new List<object>(); } foreach (DataRow dr in dt.Rows) { foreach (DataColumn col in dt.Columns) { string columnName = col.ColumnName; if (col.DataType.Name == "DateTime"&&dr[columnName].ToString().StartsWith("0")) { dr[columnName] = new DateTime(1900, 1, 1); } valueDict[columnName].Add(dr[columnName]); } } foreach (DataColumn col in dt.Columns) { string columnName = col.ColumnName; columns += ":" + columnName + ","; paramDict[columnName].Value = valueDict[columnName].ToArray(); } columns = columns.TrimEnd(‘,‘); command.CommandText = $"insert into {targetTable} values({columns})"; conn.Open(); command.ExecuteNonQuery(); } catch (Exception ex) { throw; } finally { conn.Dispose(); } } } public override void BulkCopyMappingsAdd(object copy, string source, string des) { var bulkcopy = copy as Oracle.DataAccess.Client.OracleBulkCopy; bulkcopy.ColumnMappings.Add(source, des); } public override object BuldCopy(object cn) { object copy = new Oracle.DataAccess.Client.OracleBulkCopy(cn as Oracle.DataAccess.Client.OracleConnection); return copy; } } }
以上是关于OracleBulkCopy的主要内容，如果未能解决你的问题，请参考以下文章
如何通过 oracleBulkCopy (ODP.NET) 检查插入的行数