数据库调优过程（四）：提高数据库写入修改方案

目前先后做了以下调整：

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.xxxxFTP2xx20160513xxTxxx.xml.gz时抛出异常
事务(进程 ID 69)与另一个进程被死锁在 锁 资源上，并且已被选作死锁牺牲品。请重新运行该事务。

后来发现从数据方面来解决希望不大，从程序的角度做了一下调整：

/****** Object:  StoredProcedure [dbo].[GetGlobalMaxOID]    Script Date: 07/13/2016 23:50:04 ******/
DROP PROCEDURE [dbo].[GetGlobalMaxOID]
GO
/****** Object:  Table [dbo].[GlobalMaxOID]    Script Date: 07/13/2016 23:50:03 ******/
DROP TABLE [dbo].[GlobalMaxOID]
GO
/****** Object:  Table [dbo].[GlobalMaxOID]    Script Date: 07/13/2016 23:50:03 ******/
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE TABLE [dbo].[GlobalMaxOID](
    [ENodeBID] [int] NOT NULL,
    [TableName] [nvarchar](64) NOT NULL,
    [MaxOID] [bigint] NOT NULL,
    [Flag] [int] NOT NULL
) ON [PRIMARY]
GO
/****** Object:  StoredProcedure [dbo].[GetGlobalMaxOID]    Script Date: 07/13/2016 23:50:04 ******/
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
-- =============================================
-- Author:        tommy duan
-- Create date: 2016-07-12
-- Description: OID 占坑
-- =============================================
CREATE PROCEDURE [dbo].[GetGlobalMaxOID]
    @ENodeBID int,
    @TableName nvarchar(64),
    @TempCount int
AS
BEGIN
    SET NOCOUNT ON;
    
    Declare @MaxOID bigint;
    
    Set XACT_ABORT ON;
    Begin Transaction    
        if not exists(Select ENodeBID From dbo.GlobalMaxOID Where ENodeBID=@ENodeBID and TableName=@TableName) begin
            insert into dbo.GlobalMaxOID(ENodeBID,TableName,MaxOID,Flag)VALUES(@ENodeBID,@TableName,0,0);
        end
        
        Update dbo.GlobalMaxOID Set Flag=1 Where ENodeBID=@ENodeBID and TableName=@TableName;
    
        Select @MaxOID=MaxOID From dbo.GlobalMaxOID Where ENodeBID=@ENodeBID and TableName=@TableName;
                    
        Update dbo.GlobalMaxOID Set MaxOID=(@MaxOID+@TempCount),Flag=0 Where ENodeBID=@ENodeBID and TableName=@TableName;
    Commit Transaction;    
    
    Select @MaxOID;    
END
GO

使用lock：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Data.SqlClient;
using System.Threading.Tasks;
using System.Configuration;

namespace TestGlobalMaxOID
{
    class Task
    {
        public int ENodeBID { get; set; }
        public int TempCount { get; set; }
        public string ConnectionString { get; set; }
    }

    class Program
    {
        static object locker = new object();

        static void Main(string[] args)
        {
            string tableName = string.Empty;

            Parallel.For(10000, 10100, new ParallelOptions() { MaxDegreeOfParallelism = Environment.ProcessorCount }, (int enodebid) =>
            {
                List<Task> tasks = new List<Task>();
                for (int i = 0; i < 100; i++)
                {
                    tasks.Add(new Task() { ENodeBID = enodebid, TempCount = 1000000, ConnectionString = ConfigurationManager.ConnectionStrings["ConnectionString"].ConnectionString });
                }

                Parallel.ForEach(tasks, new ParallelOptions() { MaxDegreeOfParallelism = Environment.ProcessorCount }, (Task task) =>
                {
                    Console.WriteLine(GetMaxOIDByTableNameAndENodeBID(task.ConnectionString, "MRO", task.ENodeBID, task.TempCount));
                });
            });


            Console.WriteLine("Complete!");
            Console.ReadKey();

        }

        static long GetMaxOIDByTableNameAndENodeBID(string connectionString, string tableName, int eNodeBId, int tempCount)
        {
            lock (locker)
            {
                using (SqlConnection connection = new SqlConnection(connectionString))
                {
                    using (SqlCommand command = new SqlCommand())
                    {
                        connection.Open();

                        command.Connection = connection;
                        command.CommandText = "[dbo].[GetGlobalMaxOID]";
                        command.CommandType = System.Data.CommandType.StoredProcedure;

                        command.Parameters.Add(new SqlParameter("@ENodeBID", eNodeBId));
                        command.Parameters.Add(new SqlParameter("@TableName", tableName));
                        command.Parameters.Add(new SqlParameter("@TempCount", tempCount));

                        object obj = command.ExecuteScalar();

                        return Convert.ToInt64(obj);
                    }
                }
            }
        }

    }
}