久久久久久综合网天天,极品少妇无套内射视频,日韩国产精品无码一区二区三区

分享改進高性能數據同步工具(一)

2011-11-04 15:08 熬夜的蟲子閱讀(4491) 評論(9) 收藏舉報

題外：在博文索引中暫時列出了開源的計劃一覽，蟲子開源的目的是希望能有更多的交流，部分軟件可能小得連開源協議的認證價值都沒有。不管程序有多小多簡單，用心把一個完整的設計思路、實現過程以及測試結果展現給大家。歡迎大牛拍磚，小牛問路。

蟲子的博文索引http://www.rzrgm.cn/dubing/archive/2011/11/03/2234599.html

軟件背景

拿本次高性能數據同步工具來說，目前還處于開發階段，大概是1/4的樣子。為了避免模糊，就先把這1/4分享給大家。

數據作為系統的核心價值，因為其流動性所以經常會有載體的變更。如何高性能、安全的將數據搬移是一個大家經常接觸也一直在用的課題。如果只是sql to sql可能作為程序員而言，DBA更適合這個內容，例如dts導入等。但是更多的實際場景下，可能會有文件、服務、甚至其他類型的數據流來源。所以作為碼農，我們不妨多了解一下這方面的內容。

設計思路

暫時開源程序中只做了sql to sql的一部分。直接就以這個開始來講吧。

首先是入參和返參的設計

/// <summary>
    /// 入參接口
    /// </summary>
    public interface IAOPParam
    {
         /// <summary>
         /// 目標地址
         /// </summary>
         string T_ConnectionString { get;  }

         /// <summary>
         /// 請求行數
         /// </summary>
         long MaxSize { get; }

         /// <summary>
         /// 表名
         /// </summary>
         string TableName { get; }

         /// <summary>
         /// 當前行數
         /// </summary>
         long CurrentSize { get; }

         /// <summary>
         /// 域名
         /// </summary>
         string p_Domain { get;  }

         /// <summary>
         /// 斷點文件地址
         /// </summary>
         string p_InitPath { get;  }

         /// <summary>
         /// 斷點時間
         /// </summary>
         DateTime p_Previous { get;  }

         /// <summary>
         /// 是否結束
         /// </summary>
         bool p_IsEnd { get;  }

         /// <summary>
         /// 排序方式
         /// </summary>
         string SortName { get; set; }

         /// <summary>
         /// 單次請求大小
         /// </summary>
         long SingleSize { get; }

         /// <summary>
         /// 排序主鍵
         /// </summary>
         string Sortkey { get;  }
         
         /// <summary>
         /// 是否支持事務
         /// </summary>
         bool IsTransaction { get;  }

         /// <summary>
         /// true為支持斷點 發生斷點或異常后程序終止   false為不支持斷點 遇到斷點或異常繼續填充直到此次請求完成
         /// </summary>
         bool IsBreakPoints { get;  }

         /// <summary>
         /// guid
         /// </summary>
         string T_Guid { get; }
    }

    /// <summary>
    /// 對象處理返回的入參接口（泛型）
    /// </summary>
    public interface IAOPParam<T> : IAOPParam
    {
        /// <summary>
        /// 泛型附加對象
        /// </summary>
        T ParamAttachObjectEx { get; }
    }

這樣設計的目的是考慮到服務器的內存與資源占用問題，如果數據來源的體積過大，我們將會對請求的來源進行分塊處理。另外通過排序字段或者自定義的sql語句或者存儲過程(暫未補充)可以對數據源進行高級過濾，斷點續傳的設計目前比較簡單，web程序的話植入cookie、控制臺或者cs程序通過文本媒介json格式來控制。

 #region IAOPResult

    /// <summary>
    /// 對象處理返回的結果接口
    /// </summary>
    /// <remarks>
    /// 建議在代碼調用返回值中都采用此類實例為返回值<br />
    /// 一般ResultNo小于0表示異常，0表示成功，大于0表示其它一般提示信息
    /// </remarks>
    public interface IAOPResult
    {
        /// <summary>
        /// 返回代碼
        /// </summary>
        int ResultNo { get; }

        /// <summary>
        /// 對應的描述信息
        /// </summary>
        string ResultDescription { get; }

        /// <summary>
        /// 相應的附加信息
        /// </summary>
        object ResultAttachObject { get; }

        /// <summary>
        /// 內部AOPResult
        /// </summary>
        IAOPResult InnerAOPResult { get; }

        /// <summary>
        /// 處理結果是否成功（ResultNo == 0）
        /// </summary>
        bool IsSuccess { get; }

        /// <summary>
        /// 處理結果是否失敗（ResultNo != 0 ）
        /// </summary>
        bool IsNotSuccess { get; }

        /// <summary>
        /// 處理結果是否失敗（ResultNo < 0 ）
        /// </summary>
        bool IsFailed { get; }

        /// <summary>
        /// 已處理，但有不致命的錯誤（ResultNo > 0）
        /// </summary>
        bool IsPassedButFailed { get; }

        /// <summary>
        /// 如果處理失敗，則拋出異常 
        /// </summary>
        /// <returns>返回本身</returns>
        IAOPResult ThrowErrorOnFailed();
    }

    #endregion IAOPResult

    #region IAOPResult<T>

    /// <summary>
    /// 對象處理返回的結果接口（泛型）
    /// </summary>
    public interface IAOPResult<T> : IAOPResult
    {
        /// <summary>
        /// 泛型附加對象
        /// </summary>
        T ResultAttachObjectEx { get; }
    }

    #endregion

返參的設計比較通用化，大家可以自己摸索下。自己也可以補充添加。

    /// <summary>
    /// 異常模塊異常，框架的基礎異常類，所有的異常請從本類派生
    /// </summary>
    [Serializable]
    public class BaseException : ApplicationException
    {
        /// <summary>
        /// 構造函數
        /// </summary>
        public BaseException()
            : this(0, null, null)
        {
        }

        /// <summary>
        /// 構造函數
        /// </summary>
        /// <param name="message">異常消息</param>
        /// <param name="innerException">內部異常</param>
        public BaseException(string message, Exception innerException)
            : this(0, message, innerException)
        {
        }

        /// <summary>
        /// 構造函數
        /// </summary>
        /// <param name="message">異常消息</param>
        public BaseException(string message)
            : this(0, message)
        {
        }

        /// <summary>
        /// 構造函數
        /// </summary>
        /// <param name="errorNo">異常編號</param>
        /// <param name="message">異常消息</param>
        public BaseException(int errorNo, string message)
            : this(errorNo, message, null)
        {
        }

        /// <summary>
        /// 構造函數
        /// </summary>
        /// <param name="errorNo">異常編號</param>
        /// <param name="message">異常消息</param>
        /// <param name="innerException">內部異常</param>
        public BaseException(int errorNo, string message, Exception innerException)
            : base(message, innerException)
        {
            this.errorNo = errorNo;
        }

        /// <summary>
        /// 異常編號
        /// </summary>
        protected int errorNo;

        /// <summary>
        /// 異常編號
        /// </summary>
        public int ErrorNo
        {
            get { return errorNo; }
        }

        /// <summary>
        /// 查找原始的異常
        /// </summary>
        /// <param name="e">異常</param>
        /// <returns>原始的異常</returns>
        public static Exception FindSourceException(Exception e)
        {
            Exception e1 = e;
            while (e1 != null)
            {
                e = e1;
                e1 = e1.InnerException;
            }
            return e;
        }

        /// <summary>
        /// 從異常樹種查找指定類型的異常
        /// </summary>
        /// <param name="e">異常</param>
        /// <param name="expectedExceptionType">期待的異常類型</param>
        /// <returns>所要求的異常，如果找不到，返回null</returns>
        public static Exception FindSourceException(Exception e, Type expectedExceptionType)
        {
            while (e != null)
            {
                if (e.GetType() == expectedExceptionType)
                {
                    return e;
                }
                e = e.InnerException;
            }
            return null;
        }
    }

異常基類。

    public class Log4N
    {
        public static void WarnLog(string msg)
        {
            string dateTimeStr = DateTime.Now.ToString();

            log4net.LogManager.GetLogger("WarnLog").Warn(dateTimeStr + "$$" + msg + "\r\n----------------------------------------------------------------------------");
        }

        public static void WarnLog(string msg, Exception ex)
        {
            string dateTimeStr = DateTime.Now.ToString();

            log4net.LogManager.GetLogger("WarnLog").Warn(dateTimeStr + "$$" + msg+"\r\n----------------------------------------------------------------------------", ex);
        }

        public static void DebugLog(string msg)
        {
            string dateTimeStr = DateTime.Now.ToString();

            log4net.LogManager.GetLogger("DebugLog").Debug(dateTimeStr + "$$" + msg + "\r\n----------------------------------------------------------------------------");
        }

        public static void DebugLog(string msg, Exception ex)
        {
            string dateTimeStr = DateTime.Now.ToString();

            log4net.LogManager.GetLogger("DebugLog").Debug(dateTimeStr + "$$" + msg + "\r\n----------------------------------------------------------------------------", ex);
        }

        public static void InfoLog(string msg)
        {
            string dateTimeStr = DateTime.Now.ToString();

            log4net.LogManager.GetLogger("InfoLog").Info(dateTimeStr + "$$" + msg + "\r\n----------------------------------------------------------------------------");
        }

        public static void InfoLog(string msg,IAOPParam app)
        {
            string dateTimeStr = DateTime.Now.ToString();
            var s = new StringBuilder();
            s.Append(dateTimeStr+"\r\n");
            s.Append(msg + "\r\n");
            s.AppendFormat("目標地址:" + app.T_ConnectionString + "\r\n");
            s.AppendFormat("請求行數:" + app.MaxSize + "\r\n");
            s.AppendFormat("當前行數:" + app.CurrentSize + "\r\n");
            s.AppendFormat("域名信息:" + app.p_Domain + "\r\n");
            s.AppendFormat("斷點信息地址:" + app.p_InitPath + "\r\n");
            s.AppendFormat("是否完成:" + app.p_IsEnd.ToString() + "\r\n");
            s.AppendFormat("斷點時間:" + app.p_Previous.ToString() + "\r\n");
            s.AppendFormat("單次行數:" + app.SingleSize.ToString() + "\r\n");
            s.AppendFormat("排序主鍵:" + app.Sortkey + "\r\n");
            s.AppendFormat("排序方式:" + app.SortName + "\r\n");
            s.AppendFormat("表名:" + app.TableName + "\r\n");

            log4net.LogManager.GetLogger("InfoLog").Info("$$" + s + "\r\n----------------------------------------------------------------------------");
        }

        public static void WarnLog(string msg, IAOPParam app)
        {
            string dateTimeStr = DateTime.Now.ToString();
            var s = new StringBuilder();
            s.Append(dateTimeStr+"\r\n");
            s.Append(msg + "\r\n");
            s.AppendFormat("目標地址:" + app.T_ConnectionString + "\r\n");
            s.AppendFormat("請求行數:" + app.MaxSize + "\r\n");
            s.AppendFormat("當前行數:" + app.CurrentSize + "\r\n");
            s.AppendFormat("域名信息:" + app.p_Domain + "\r\n");
            s.AppendFormat("斷點信息地址:" + app.p_InitPath + "\r\n");
            s.AppendFormat("是否完成:" + app.p_IsEnd.ToString() + "\r\n");
            s.AppendFormat("斷點時間:" + app.p_Previous.ToString() + "\r\n");
            s.AppendFormat("單次行數:" + app.SingleSize.ToString() + "\r\n");
            s.AppendFormat("排序主鍵:" + app.Sortkey + "\r\n");
            s.AppendFormat("排序方式:" + app.SortName + "\r\n");
            s.AppendFormat("表名:" + app.TableName + "\r\n");

            log4net.LogManager.GetLogger("WarnLog").Info("$$" + s + "\r\n----------------------------------------------------------------------------");
        }
    }

日志采取lognet 不贅述

 public static class Singleton<T> where T : class, new()
    {
       
        private static readonly object LockKey = new object();

        public static T GetInstance()
        {
            return GetInstance(null);
        }

        public static T GetInstance(Func<T> onCreateInstance)
        {
            if (_instance == null)
            {
                lock (LockKey)
                {
                    if (_instance == null)
                    {
                        try
                        {
                            return _instance = onCreateInstance == null ? new T() : onCreateInstance();
                        }
                        catch
                        {
                            _instance = default(T);
                        }
                    }
                }
            }
            return _instance;
        }
        private static T _instance;
        
        public static T GetInstance(object lockKey, T instance, Func<T> onCreateInstance)
        {
            if (instance == null)
            {
                if (lockKey == null)
                    lockKey = LockKey;
                lock (lockKey)
                {
                    try
                    {
                        instance = onCreateInstance == null ? new T() : onCreateInstance();
                    }
                    catch
                    {
                        instance = default(T);
                    }
                }
            }
            return instance;
        }

        public static void ReleaseInstance()
        {
            lock (LockKey)
            {
                var id = _instance as IDisposable;
                if (id != null)
                    id.Dispose();

                _instance = default(T);
            }
        }
     
        public static void TakeAction(Func<bool> lockCondition, object lockObject, Action action)
        {
            if (lockCondition())
            {
                lock (lockObject)
                {
                    if (lockCondition())
                    {
                        action();
                    }
                }
            }
        }
    }

單例通用類關于作用可以參考蟲子設計模式隨筆中的相關博文

public class Status : IDisposable
    {
        private readonly HttpContext _mHttpContext;
        public enum CopyStatus { Initialization, Doing, Finished }
        public string MGuid;
        public DateTime MStartTime;
        public long MTotalSize;
        public long MCurrentSize;
        public CopyStatus Statusflag;

        public Status()
		{
            MGuid = Guid.NewGuid().ToString();
			MStartTime = DateTime.Now;
			_mHttpContext = HttpContext.Current;

            Statusflag = CopyStatus.Initialization;
			MCurrentSize = 0;			
		}

        public void Dispose()
        {         
            if (_mHttpContext.Session[MGuid] != null)
            {
                _mHttpContext.Session.Remove(MGuid);
            }
        }
    }

狀態類，通過這個類可以反映出當前數據同步的進度。

邊緣化的準備工作大體如此，下面是主要的實現過程。過程中有幾個注意點，同步讀寫還是異步讀寫、是否存在線程安全甚至進程的資源安全（例如我在讀寫前5000條的時候突然在另外一個客戶端CRUD了N條數據）,另外，我們讀寫的時候是用連接的方式還是使用非連接的方式，如何解決服務器端內存占用問題，如何實現excel、txt、sql、oracle等不同數據來源的多態性。

實現過程

這里就先介紹下已經解決的一些問題

    public class AnalyseDataManager
    {
        public Status MStatus { get; set; }
        public static int SingleSize = 5000;
        public static int StatusSize = 1000;
        readonly Sqlhelper _sh = new Sqlhelper();

        public AnalyseDataManager()
        {
        }

        public AnalyseDataManager(Status st)
            : this()
        {
            MStatus = st;
        }

        public IAsyncResult OutMethod(AopParam app)
        {
            MStatus.MTotalSize = app.MaxSize;
            var func = new Func<AopParam, bool>(ServerMethod);
            return func.BeginInvoke(app, CallbackMethod, func);
        }

        /// <summary>
        /// 復制數據
        /// </summary>
        /// <returns>是否成功</returns>
        public bool ServerMethod(AopParam app)
        {
            try
            {
                _sh.App = app;
                if (_sh.OpenConn().IsSuccess)
                {
                    while (app.MaxSize > MStatus.MCurrentSize)
                    {
                        app.CurrentSize = MStatus.MCurrentSize;
                        if (!AsyncDataToServer(app) && app.IsBreakPoints)
                        {
                            break;
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                Log4N.WarnLog("ServerMethod出錯", ex);
                if (app.IsBreakPoints)
                {
                    return false;
                }
            }
            finally
            {
                _sh.Dispose();
            }
            return true;
        }

        private bool AsyncDataToServer(AopParam app)
        {
            Log4N.InfoLog(string.Format("數據同步開始\r\n來源數據{0}\r\n表的名字{1}\r\n一次性提交的行數{2}\r\n當前行數{3}", app.T_ConnectionString, app.TableName, app.MaxSize, app.CurrentSize));

            using (var bcp = new SqlBulkCopy(_sh.TconnSql))
            {
                MStatus.Statusflag = Status.CopyStatus.Doing;
                bcp.BatchSize = SingleSize;
                bcp.DestinationTableName = app.TableName;

                bcp.SqlRowsCopied +=
                  OnSqlRowsCopied;
                bcp.NotifyAfter = StatusSize;

                try
                {
                    bcp.WriteToServer(_sh.GetDtResultImp());
                }
                catch (Exception ex)
                {
                    Log4N.WarnLog("AsyncDataToServer出錯", ex);
                    return false;
                }
                finally
                {
                    _sh.IreaderSql.Close();
                }

                return true;

            }
        }
        private void OnSqlRowsCopied(object sender, SqlRowsCopiedEventArgs e)
        {
            Thread.Sleep(1000);
            MStatus.MCurrentSize += StatusSize;
        }

        public void CallbackMethod(IAsyncResult ar)
        {
            var caller = (Func<AopParam,bool >)ar.AsyncState;
            if (caller.EndInvoke(ar))
            {
                MStatus.Statusflag = Status.CopyStatus.Finished;
            }
        }

    }

Microsoft SQL Server 提供一個稱為 bcp 的流行的命令提示符實用工具，用于將數據從一個表移動到另一個表（表既可以在同一個服務器上，也可以在不同服務器上）。SqlBulkCopy 類允許編寫提供類似功能的托管代碼解決方案。還有其他將數據加載到 SQL Server 表的方法（例如 INSERT 語句），但相比之下 SqlBulkCopy 提供明顯的性能優勢。使用 SqlBulkCopy 類只能向 SQL Server 表寫入數據。但是，數據源不限于 SQL Server；可以使用任何數據源，只要數據可加載到 DataTable 實例或可使用 IDataReader 實例讀取數據。其中 SqlRowsCopied 在每次處理完 NotifyAfter 屬性指定的行數時發生。

ServerMethod為主方法提供單次客戶端請求的邏輯。

OutMethod對外開放以bpm異步編程模型形式進行處理、sqlhelper之所以不設計成單列，為了保證可以多個客戶端請求狀態不干擾。

 public class Sqlhelper : IDisposable
    {
        private readonly string _sqlconn = ConfigurationSettings.AppSettings["BaseConn"];
        public bool IblnTransBegin { get; set; }
        public SqlTransaction ItransSql { get; set; }
        public SqlConnection IconnSql { get; set; }
        public SqlConnection TconnSql { get; set; }
        public SqlDataReader IreaderSql { get; set; }
        public IAOPParam App { get; set; }

        public DataTable GetDtResult(string sqlcommand)
        {
            var ds = new DataSet();
            var da = new SqlDataAdapter(sqlcommand, new SqlConnection(_sqlconn));
            da.Fill(ds);
            if (ds.Tables[0] != null)
            {
                return ds.Tables[0];
            }
            return null;
        }

        public DataTable GetDtResult()
        {
            //string sqlstr = string.Format("Select Top {0} * From {1} Where {2} not in (select Top {4} {2} From {1} order by {2} {3} ) order by {2} {3}  ", app.SingleSize.ToString(), app.TableName, app.Sortkey, app.SortName, app.CurrentSize.ToString());
            string sqlstr = GetCommandByApp();
            var ds = new DataSet();
            var da = new SqlDataAdapter(sqlstr, new SqlConnection(_sqlconn));
            da.Fill(ds);
            if (ds.Tables[0] != null)
            {
                return ds.Tables[0];
            }
            return null;
        }

        public SqlDataReader GetDtResultImp()
        {
            var sqlstr = GetCommandByApp();
            var command = new SqlCommand(
               sqlstr, IconnSql);
            IreaderSql =
              command.ExecuteReader();
            return IreaderSql;
        }

        public IAOPResult OpenConn()
        {
            var ar = new AOPResult(0);
            IconnSql = new SqlConnection(_sqlconn);
            TconnSql = new SqlConnection(App.T_ConnectionString);
            try
            {
                IconnSql.Open();
                TconnSql.Open();
            }
            catch (SqlException ex)
            {
                ar.ResultNo = 1;
                Log4N.InfoLog(string.Format("OpenConn失敗,詳細消息為{0},源表", ex.Message), App);
            }
            return ar;
        }

        public IAOPResult CloseConn()
        {

            var ar = new AOPResult(0);
            try
            {
                IconnSql.Close();
                TconnSql.Close();
            }
            catch (SqlException ex)
            {
                ar.ResultNo = 1;
                Log4N.InfoLog(string.Format("CloseConn失敗,詳細消息為{0},源表", ex.Message), App);
            }
            return ar;
        }

        public IAOPResult BeginTran()
        {
            ItransSql = IconnSql.BeginTransaction();
            return null;
        }

        public void Dispose()
        {
            CloseConn();
        }

        public string GetCommandByApp()
        {
            string sqlstr = string.Empty;
            if(App.CurrentSize == 0)
            {
                switch (App.SortName.ToLower())
                {
                    case "asc":
                        sqlstr = string.Format("Select Top {0} * From {1}  order by {2} asc", App.SingleSize.ToString(), App.TableName, App.Sortkey);
                        break;
                    case "desc":
                        sqlstr = string.Format("Select Top {0} * From {1}  order by {2} desc", App.SingleSize.ToString(), App.TableName, App.Sortkey);
                        break;
                }
            }
            else
            {
                switch (App.SortName.ToLower())
                {
                    case "asc":
                        sqlstr = string.Format("Select Top {0} * From {1} Where {2} >(select max ({2}) From (select Top {3} {2} From {1} order by {2} asc ) as temp_chongzi) order by {2} asc", App.SingleSize.ToString(), App.TableName, App.Sortkey, App.CurrentSize.ToString());
                        break;
                    case "desc":
                        sqlstr = string.Format("Select Top {0} * From {1} Where {2} <(select min ({2}) From (select Top {3} {2} From {1}) order by {2} desc )as temp_chongzi) order by {2} desc", App.SingleSize.ToString(), App.TableName, App.Sortkey, App.CurrentSize.ToString());
                        break;
                }      

            }
            return sqlstr;

        }

數據庫訪問層中首先是一個類似分頁sql的設計，來優化單次請求的效率。bcp的來源可以選擇連接式的SqlDataReader 或者非連接式的Dataset，2者各有優缺。前者需要打開SqlConnection，但是是逐條讀取，后者非連接但是占用內存大。至于具體的性能比，蟲子在下一章節再和大家討論。至于源程序目前還是草稿版，很多功能還未實現，細節處理也不夠細膩，因為異步目前只設置了一個線程，還未涉及到并行框架，性能方面還有相當大的提高空間。先放出來讓大家討論，細節方面可以暫時先略過，大家可以說說在設計方面如何才能更高效、穩定。

源碼地址：點擊此處下載

刷新頁面返回頂部

熬夜的蟲子

分享改進 高性能數據同步工具(一)

About

分享改進高性能數據同步工具(一)