目前先后做了以下调整：
1、单表多个线程写入的时候，总会因为写入log或单文件写入速度的限制，导致IO提到不了成为入库的瓶颈；为了提到入库IO速度，我们采取分表策略，一次性拆分到一个ENB为单元的物理表；
2、修改批量入库方案：从监控IO来看，IO的写入速度确实有提高，但是同时写入多个表过程是单线程批量写入单个表，后来修改为并行写入多个表。但是依然存在内存居高不下的情况，这样的一个enb往往会造成内存居高不下，以至于内存资源成为瓶颈；
3、为了解决内存成为瓶颈，在内存处理时，每当表记录到达50w，就批量入库一次，以便用来清理内存信息；
4、基于以上方案，如果集中遇到大文件时，依然会出现内存不足，一个enb读取解析，入库都占用时长过长，而有小文件时，就不会出现占用时间过长的问题，所以针对任务采取控制同时运行的任务大文件不超过两个；
5、之前一次读取enb的下的所有文件，拆分为一次性只读取一个文件，就解析入库，缩短每个线程占用时长，避免因为单个线程占用时长过长进而对cpu调度造成压力，避免线程给后置的情况发生；而且改为小文件为一个单独线程来处理后，每个线程占用的内存会降低，会增加同时入库同一个物理表的线程数增加，进而有可能提到IO利用率，进而提高入库性能。
拆分为一个文件已入库，就会涉及到并行入库同一个或多个物理表的情况，而此时表结构是主子表结构，主表与字表之间通过一个oid来关联，必须要维护这个oid，以保证主表与字表之间的关联关系，如果并行插入单个表就常常会出现同一个oid已被占用的异常情况，我们采用一个GlobalMaxOID,来占坑每个物理表已经被占用的OID。
在处理OID占坑策略时遇到了一下问题：
并行处理一个ENB内部文件\\17x.xx.x.xx\xx\FTP2\xx\20160513\xx\Txxx.xml.gz时抛出异常
事务(进程 ID 69)与另一个进程被死锁在 锁 资源上，并且已被选作死锁牺牲品。请重新运行该事务。
后来发现从数据方面来解决希望不大，从程序的角度做了一下调整：
1 /****** Object: StoredProcedure [dbo].[GetGlobalMaxOID] Script Date: 07/13/2016 23:50:04 ******/ 2 DROP PROCEDURE [dbo].[GetGlobalMaxOID] 3 GO 4 /****** Object: Table [dbo].[GlobalMaxOID] Script Date: 07/13/2016 23:50:03 ******/ 5 DROP TABLE [dbo].[GlobalMaxOID] 6 GO 7 /****** Object: Table [dbo].[GlobalMaxOID] Script Date: 07/13/2016 23:50:03 ******/ 8 SET ANSI_NULLS ON 9 GO 10 SET QUOTED_IDENTIFIER ON 11 GO 12 CREATE TABLE [dbo].[GlobalMaxOID]( 13 [ENodeBID] [int] NOT NULL, 14 [TableName] [nvarchar](64) NOT NULL, 15 [MaxOID] [bigint] NOT NULL, 16 [Flag] [int] NOT NULL 17 ) ON [PRIMARY] 18 GO 19 /****** Object: StoredProcedure [dbo].[GetGlobalMaxOID] Script Date: 07/13/2016 23:50:04 ******/ 20 SET ANSI_NULLS ON 21 GO 22 SET QUOTED_IDENTIFIER ON 23 GO 24 -- ============================================= 25 -- Author: tommy duan 26 -- Create date: 2016-07-12 27 -- Description: OID 占坑 28 -- ============================================= 29 CREATE PROCEDURE [dbo].[GetGlobalMaxOID] 30 @ENodeBID int, 31 @TableName nvarchar(64), 32 @TempCount int 33 AS 34 BEGIN 35 SET NOCOUNT ON; 36 37 Declare @MaxOID bigint; 38 39 Set XACT_ABORT ON; 40 Begin Transaction 41 if not exists(Select ENodeBID From dbo.GlobalMaxOID Where ENodeBID=@ENodeBID and TableName=@TableName) begin 42 insert into dbo.GlobalMaxOID(ENodeBID,TableName,MaxOID,Flag)VALUES(@ENodeBID,@TableName,0,0); 43 end 44 45 Update dbo.GlobalMaxOID Set Flag=1 Where ENodeBID=@ENodeBID and TableName=@TableName; 46 47 Select @MaxOID=MaxOID From dbo.GlobalMaxOID Where ENodeBID=@ENodeBID and TableName=@TableName; 48 49 Update dbo.GlobalMaxOID Set MaxOID=(@MaxOID+@TempCount),Flag=0 Where ENodeBID=@ENodeBID and TableName=@TableName; 50 Commit Transaction; 51 52 Select @MaxOID; 53 END 54 GO
使用lock：
1 using System;
2 using System.Collections.Generic;
3 using System.Linq;
4 using System.Text;
5 using System.Data.SqlClient;
6 using System.Threading.Tasks;
7 using System.Configuration;
8
9 namespace TestGlobalMaxOID
10 {
11 class Task
12 {
13 public int ENodeBID { get; set; }
14 public int TempCount { get; set; }
15 public string ConnectionString { get; set; }
16 }
17
18 class Program
19 {
20 static object locker = new object();
21
22 static void Main(string[] args)
23 {
24 string tableName = string.Empty;
25
26 Parallel.For(10000, 10100, new ParallelOptions() { MaxDegreeOfParallelism = Environment.ProcessorCount }, (int enodebid) =>
27 {
28 List<Task> tasks = new List<Task>();
29 for (int i = 0; i < 100; i++)
30 {
31 tasks.Add(new Task() { ENodeBID = enodebid, TempCount = 1000000, ConnectionString = ConfigurationManager.ConnectionStrings["ConnectionString"].ConnectionString });
32 }
33
34 Parallel.ForEach(tasks, new ParallelOptions() { MaxDegreeOfParallelism = Environment.ProcessorCount }, (Task task) =>
35 {
36 Console.WriteLine(GetMaxOIDByTableNameAndENodeBID(task.ConnectionString, "MRO", task.ENodeBID, task.TempCount));
37 });
38 });
39
40
41 Console.WriteLine("Complete!");
42 Console.ReadKey();
43
44 }
45
46 static long GetMaxOIDByTableNameAndENodeBID(string connectionString, string tableName, int eNodeBId, int tempCount)
47 {
48 lock (locker)
49 {
50 using (SqlConnection connection = new SqlConnection(connectionString))
51 {
52 using (SqlCommand command = new SqlCommand())
53 {
54 connection.Open();
55
56 command.Connection = connection;
57 command.CommandText = "[dbo].[GetGlobalMaxOID]";
58 command.CommandType = System.Data.CommandType.StoredProcedure;
59
60 command.Parameters.Add(new SqlParameter("@ENodeBID", eNodeBId));
61 command.Parameters.Add(new SqlParameter("@TableName", tableName));
62 command.Parameters.Add(new SqlParameter("@TempCount", tempCount));
63
64 object obj = command.ExecuteScalar();
65
66 return Convert.ToInt64(obj);
67 }
68 }
69 }
70 }
71
72 }
73 }
时间： 2024-12-19 17:35:19