程序員修仙之路--突破內存限制的高性能排序

菜菜的漲工資申請還在待審批中....

做爲一個技術人員，技術的問題仍是要解決。通過線上日誌的分析，日誌採用小時機制，一個小時一個日誌文件，同一個小時的日誌文件有多個，也就是說同一時間內的日誌有可能分散在多個日誌文件中，這也是Y總要合併的主要緣由。每一個日誌文件大約有500M，大約有100個。此時，若是你閱讀到此文章，該怎麼作呢？不如先靜心想2分鐘！！

問題分析

要想實現Y總的需求其實仍是有幾個難點的：

1. 如何能把全部的日誌文件按照時間排序
2. 日誌文件的總大小爲500M*100 ，大約50G，因此所有加載到內存是不可能的
3. 程序執行過程當中，要頻繁排序並查找最小元素。

那咱們該怎麼作呢？其中一個解決方案就是它：堆

解決方案

堆定義

堆（英語：heap)是計算機科學中一類特殊的數據結構的統稱。堆一般是一個能夠被看作一棵樹的數組對象。堆老是知足下列性質：

• 堆中某個節點的值老是不大於或不小於其父節點的值
• 堆老是一棵徹底二叉樹（徹底二叉樹要求，除了最後一層，其餘層的節點個數都是滿的，最後一層的節點都靠左排列）

對於每一個節點的值都大於等於子樹中每一個節點值的堆，咱們叫做「大頂堆」。對於每一個節點的值都小於等於子樹中每一個節點值的堆，咱們叫做「小頂堆」。

堆實現

徹底二叉樹比較適合用數組來存儲（鏈表也能夠實現）。爲何這麼說呢？用數組來存儲徹底二叉樹是很是節省存儲空間的。由於咱們不須要存儲左右子節點的指針，單純地經過數組的下標，就能夠找到一個節點的左右子節點和父節點。

通過上圖能夠發現，數組位置0爲空，雖然浪費了一個存儲空間，可是當計算元素在數組位置的時候確很是方便：數組下標爲X的元素的左子樹的下標爲2x，右子樹的下標爲2x+1。
其實實現一個堆很是簡單，就是順着元素所在的路徑，向上或者向下對比而後交換位置。

1. 添加元素

添加元素的時候咱們習慣採用自下而上的調整方式來調整堆，咱們在數組的最後一個空閒位置插入新元素，按照堆的下標上標原則查找到父元素對比，若是小於父元素的值（大頂堆），則互相交換。如圖：

1. 刪除最大（最小元素）

對於大頂堆，堆頂的元素就是最大元素。刪除該元素以後，咱們須要把第二大元素提到堆頂位置。依次類推，直到把路徑上的全部元素都調整完畢。

擴展閱讀

1. 小頂堆的頂部元素其實就是整個堆最小的元素，大頂堆頂部元素是整個堆的最大元素。這也是堆排序的最大優勢，取最小元素或者最大元素時間複雜度爲O(1)
2. 刪除元素的時候咱們要注意一點，若是採用自頂向下交換元素的方式，在不少狀況下形成堆嚴重的不平衡（左右子樹深度相差較大）的狀況，爲了防止相似狀況，咱們能夠把最後一個元素提到堆頂，而後調整的策略，由於最後一個元素老是在最後一級，不會形成左右子樹相差很大的狀況。
3. 對於有重複元素的堆，一種解決方法是認爲是誰先誰大，後進入堆的元素小於先進入堆的元素，這樣在查找的時候必定要查完全才行。另一種方式是在堆的每一個元素中存儲一個鏈表，用來存放相同的元素，原理相似於散列表。不過這樣在刪除這個元素的時候須要特殊處理一下。
4. 刪除堆頂數據和往堆中插入數據的時間複雜度都是 O(logn)。
5. 不斷調整堆的過程其實就是排序過程，在某些場景下，咱們能夠利用堆來實現排序。

asp.net core 模擬代碼

如下代碼通過少量修改甚至不修改的狀況下可直接在生產環境應用

小頂堆實現代碼

/// <summary>
    /// 小頂堆,T類型須要實現 IComparable 接口
    /// </summary>
    class MinHeap<T> where T : IComparable
    {
        private T[] container; // 存放堆元素的容器
        private int capacity;  // 堆的容量，最大能夠放多少個元素
        private int count; // 堆中已經存儲的數據個數

        public MinHeap(int _capacity)
        {
            container = new T[_capacity + 1];
            capacity = _capacity;
            count = 0;
        }
        //插入一個元素
        public bool AddItem(T item)
        {
            if (count >= capacity)
            {
                return false;
            }
            ++count;
            container[count] = item;
            int i = count;
            while (i / 2 > 0 && container[i].CompareTo(container[i / 2]) < 0)
            {
                // 自下往上堆化，交換 i 和i/2 元素
                T temp = container[i];
                container[i] = container[i / 2];
                container[i / 2] = temp;
                i = i / 2;
            }
            return true;
        }
        //獲取最小的元素
        public T GetMinItem()
        {
            if (count == 0)
            {
                return default(T);
            }
            T result = container[1];
            return result;
        }
        //刪除最小的元素，即堆頂元素
        public bool DeteleMinItem()
        {
            if (count == 0)
            {
                return false;
            }
            container[1] = container[count];
            container[count] = default(T);
            --count;
            UpdateHeap(container, count, 1);
            return true;
        }
        //從某個節點開始從上向下 堆化
        private void UpdateHeap(T[] a, int n, int i)
        {
            while (true)
            {
                int maxPos = i;
                //遍歷左右子樹，肯定那個是最小的元素
                if (i * 2 <= n && a[i].CompareTo(a[i * 2]) > 0)
                {
                    maxPos = i * 2;
                }
                if (i * 2 + 1 <= n && a[maxPos].CompareTo(a[i * 2 + 1]) > 0)
                {
                    maxPos = i * 2 + 1;
                }
                if (maxPos == i)
                {
                    break;
                }
                T temp = container[i];
                container[i] = container[maxPos];
                container[maxPos] = temp;
                i = maxPos;
            }
        }
    }

模擬日誌文件內容

//由於須要不停的從log文件讀取內容，因此須要一個和log文件保持鏈接的包裝
    class LogInfoIndex : IComparable
    {
        //標誌內容來自於哪一個文件
        public int FileIndex { get; set; }
        //具體的日誌文件內容
        public LogInfo Data { get; set; }

        public int CompareTo(object obj)
        {
            var tempInfo = obj as LogInfoIndex;
            if (this.Data.Index > tempInfo.Data.Index)
            {
                return 1;
            }
            else if (this.Data.Index < tempInfo.Data.Index)
            {
                return -1;
            }
            return 0;
        }
    }
    class LogInfo
    {       
        //用int來模擬datetime 類型，由於用int 看的最直觀
        public int Index { get; set; }
        public string UserName { get; set; }
    }

生成模擬日誌程序

static void WriteFile()
        {
            int fileCount = 0;
            while (fileCount < 10)
            {
                string filePath = $@"D:\log\{fileCount}.txt";
                int index = 0;
                while (index < 100000)
                {
                    LogInfo info = new LogInfo() { Index = index, UserName = Guid.NewGuid().ToString() };
                    File.AppendAllText(filePath, JsonConvert.SerializeObject(info)+ "\r\n");
                    index++;
                }
                fileCount++;
            }
           
        }

文件內容以下：

測試程序

static void Main(string[] args)
        {
            int heapItemCount = 10;
            int startIndex = 0;
            StreamReader[] allReader = new StreamReader[10];
            MinHeap<LogInfoIndex> container = new MinHeap<LogInfoIndex>(heapItemCount);

            //首先每一個文件讀取一條信息          
            while(startIndex< heapItemCount)
            {
                string filePath = $@"D:\log\{startIndex}.txt";
                System.IO.StreamReader reader = new System.IO.StreamReader(filePath);
                allReader[startIndex] = reader;
                string content= reader.ReadLine();
                var contentObj = JsonConvert.DeserializeObject<LogInfo>(content);
                LogInfoIndex item = new LogInfoIndex() {  FileIndex= startIndex , Data= contentObj };
                container.AddItem(item);
                startIndex++;
            }
            //而後開始循環出堆，入堆
            while (true)
            {
                var heapFirstItem = container.GetMinItem();
                if (heapFirstItem == null)
                {
                    break;
                }
                container.DeteleMinItem();
                
                File.AppendAllText($@"D:\log\total.txt", JsonConvert.SerializeObject(heapFirstItem.Data) + "\r\n");
                var nextContent = allReader[heapFirstItem.FileIndex].ReadLine();
                if (string.IsNullOrWhiteSpace( nextContent))
                {
                    //若是其中一個文件已經讀取完畢 則跳過
                    continue;
                }
                var contentObj = JsonConvert.DeserializeObject<LogInfo>(nextContent);
                LogInfoIndex item = new LogInfoIndex() { FileIndex = heapFirstItem.FileIndex, Data = contentObj };
                container.AddItem(item);
            }
            //釋放StreamReader
            foreach (var reader in allReader)
            {
                reader.Dispose();
            }
            Console.WriteLine("完成");        
            Console.Read();
        }

結果以下：

---

添加關注，查看更精美版本，收穫更多精彩