Program.cs代码:
class Program { static void Main(string[] args) { var test = new PgBulkCopyHelper ("bld_amap_gzmain"); foreach (string pName in test.PropNames) { Console.WriteLine("name: {0},\t\ttype: {1}", pName, test.PropInfo[pName]); } //----------------------------------------------------------------------------------------------- //定义每次插入的最大数量限制 int maxNum = 1; //100000; //初始化对应的数据表 DataTable dataTable = test.InitDataTable(); string connectionString = "Host=localhost;Username=king;Password=wu12345;Database=dellstore"; List > bldsList = new List >(); NpgsqlPolygon plg1 = new NpgsqlPolygon(10); plg1.Add(new NpgsqlPoint(0.0, 0.0)); plg1.Add(new NpgsqlPoint(6.0, -1.0)); plg1.Add(new NpgsqlPoint(5.0, 3.0)); plg1.Add(new NpgsqlPoint(1.0, 2.0)); NpgsqlPolygon plg2 = new NpgsqlPolygon(10); plg2.Add(new NpgsqlPoint(100.0, 10.0)); plg2.Add(new NpgsqlPoint(40.0, 180.0)); plg2.Add(new NpgsqlPoint(190.0, 60.0)); plg2.Add(new NpgsqlPoint(10.0, 60.0)); plg2.Add(new NpgsqlPoint(160.0, 180.0)); List sblist1 = new List (){ new SingleBuilding(){id=System.Guid.NewGuid(), tile_x=1, tile_y=2, bps_gc=plg1, bps_llc=plg2, cp_gc=new NpgsqlPoint(0,0), cp_llc=new NpgsqlPoint(100,10), name="测试文本1", bld_floor=111, height=22 }, new SingleBuilding(){id=System.Guid.NewGuid(), tile_x=1, tile_y=2, bps_gc=plg1, bps_llc=plg2, cp_gc=new NpgsqlPoint(0,0), cp_llc=new NpgsqlPoint(100,10), name="测试文本2", bld_floor=222, height=444 } }; bldsList.Add(sblist1); using (var conn = new NpgsqlConnection(connectionString)) { conn.Open(); foreach (List blds in bldsList) { if (blds != null && blds.Count > 0) { //填充数据 test.FillDataTable(blds, dataTable); } //判断 dataTable 里面的数据量是否已经超过规定最大行数 maxNum if (dataTable.Rows.Count>maxNum) { //如果是，则将 dataTable 里面的数据插入到数据库中 test.BulkInsert(conn, dataTable); //清空 dataTable 中的现有数据 dataTable.Clear(); } } } } } public class SingleBuilding { //创建数据表的SQL语句如下： /* CREATE TABLE bld_amap_gzmain ( id uuid PRIMARY KEY NOT NULL, tile_x integer, --x index of the map tile where the building is located tile_y integer, --y index of the map tile where the building is located bps_gc polygon NOT NULL, --the points of the bottom outline of the building, geodetic coordinates bps_llc polygon NOT NULL, --the points of the bottom outline of the building, Latitude and longitude coordinates cp_gc point NOT NULL, --the center point of the building, geodetic coordinates cp_llc point NOT NULL, --the center point of the building, Latitude and longitude coordinates name text, bld_floor smallint, --the number of floors of the building height real --the height of building ); */ public Guid id { get; set; } public int? tile_x { get; set; } public int? tile_y { get; set; } public NpgsqlPolygon bps_gc { get; set; } public NpgsqlPolygon bps_llc { get; set; } public NpgsqlPoint cp_gc { get; set; } public NpgsqlPoint cp_llc { get; set; } public string name { get; set; } public short? bld_floor { get; set; } public float? height { get; set; } }
PgBulkCopyHelper.cs代码:
using Npgsql;using System;using System.Collections.Generic;using System.Data;using System.Globalization;using System.Linq;using System.Reflection;namespace PgBulkCopyHelper{ /// /// 用以快速将大批量数据插入到postgresql中 /// /// public class PgBulkCopyHelper { /// /// TEntity的属性信息 /// Dictionary(string "property_name", Type property_type) /// public Dictionary PropInfo { get; set; } /// /// TEntity的属性名称列表 /// public List PropNames { get; set; } /// /// 数据表全名：schema.tableName or tableName /// public string FullTableName { get; set; } /// /// 构造函数 /// /// 数据表的schema，一般为public /// 数据表的名称 public PgBulkCopyHelper(string schema, string tableName) { PropNames = new List (); PropInfo = new Dictionary (); PropertyInfo[] typeArgs = GetPropertyFromTEntity(); foreach (PropertyInfo tParam in typeArgs) { PropNames.Add(tParam.Name); PropInfo[tParam.Name] = tParam.PropertyType; } if (!string.IsNullOrWhiteSpace(tableName)) { if (string.IsNullOrWhiteSpace(schema)) { FullTableName = tableName; } else FullTableName = string.Format("{0}.{1}", schema, tableName); } } /// /// 构造函数 /// /// 数据表的名称 public PgBulkCopyHelper(string tableName) :this(null, tableName) { } /// /// 获取TEntity的属性信息 /// /// TEntity的属性信息的列表 private PropertyInfo[] GetPropertyFromTEntity() { Type t = typeof(TEntity); PropertyInfo[] typeArgs = t.GetProperties(); return typeArgs; } /// /// 根据TEntity的属性信息构造对应数据表 /// /// 只有字段信息的数据表 public DataTable InitDataTable() { DataTable dataTable = new DataTable(); foreach(PropertyInfo tParam in GetPropertyFromTEntity()) { Type propType = tParam.PropertyType; //由于 DataSet 不支持 System.Nullable<> 类型，因此要先做判断 if ((propType.IsGenericType) && (propType.GetGenericTypeDefinition() == typeof(Nullable<>))) propType = propType.GetGenericArguments()[0]; dataTable.Columns.Add(tParam.Name, propType); } return dataTable; } /// /// 根据TEntity可枚举列表填充给定的数据表 /// /// TEntity类型的可枚举列表 /// 数据表 public void FillDataTable(IEnumerable entities, DataTable dataTable) { if (entities != null && entities.Count() > 0) { foreach (TEntity entity in entities) { FillDataTable(entity, dataTable); } } } /// /// 在DataTable中插入单条数据 /// /// 具体数据 /// 数据表 public void FillDataTable(TEntity entity, DataTable dataTable) { var dataRow = dataTable.NewRow(); int colNum = dataTable.Columns.Count; PropertyInfo[] typeArgs = GetPropertyFromTEntity(); for (int i = 0; i < colNum; i++) { dataRow[i] = typeArgs[i].GetValue(entity); } dataTable.Rows.Add(dataRow); } /// /// 通过PostgreSQL连接把dataTable中的数据整块填充到数据库对应的数据表中 /// 注意，该函数不负责NpgsqlConnection的创建、打开以及关闭 /// /// PostgreSQL连接 /// 数据表 public void BulkInsert(NpgsqlConnection conn, DataTable dataTable) { var commandFormat = string.Format(CultureInfo.InvariantCulture, "COPY {0} FROM STDIN BINARY", FullTableName); using (var writer = conn.BeginBinaryImport(commandFormat)) { foreach (DataRow item in dataTable.Rows) writer.WriteRow(item.ItemArray); } } }} 运行结果如图:
