突破內存限制的高性能排序
菜菜呀，昨天晚上班級空間崩潰了
程序員主力 Y總
what？
菜菜
我看服務器上寫了很多個日誌文件，我看著太費勁了，能不能按照日期排序整合成一個文件呀？
程序員主力 Y總
Y總要查日誌呀？
菜菜
我就是喜歡編程，編程就是我的全部，給你半個小時搞一下
程序員主力 Y總
天天這麽短時間搞這麽多爛七八糟的需求，能不能給我漲點工資呀？
菜菜
你去和X總說，我不管這事，我只管編程！！
程序員主力 Y總
............
菜菜
菜菜的漲工資申請還在待審批中....
作為一個技術人員，技術的問題還是要解決。經過線上日誌的分析，日誌采用小時機制，一個小時一個日誌文件，同一個小時的日誌文件有多個，也就是說同一時間內的日誌有可能分散在多個日誌文件中，這也是Y總要合並的主要原因。每個日誌文件大約有500M，大約有100個。此時，如果你閱讀到此文章，該怎麽做呢？不如先靜心想2分鐘！！
問題分析
要想實現Y總的需求其實還是有幾個難點的：
1. 如何能把所有的日誌文件按照時間排序
2. 日誌文件的總大小為500M*100 ，大約50G，所以全部加載到內存是不可能的
3. 程序執行過程中，要頻繁排序並查找最小元素。
那我們該怎麽做呢？其中一個解決方案就是它：堆
解決方案堆定義
堆（英語：heap)是計算機科學中一類特殊的數據結構的統稱。堆通常是一個可以被看做一棵樹的數組對象。
堆總是滿足下列性質：
1. 堆中某個節點的值總是不大於或不小於其父節點的值
2. 堆總是一棵完全二叉樹（完全二叉樹要求，除了最後一層，其他層的節點個數都是滿的，最後一層的節點都靠左排列）
對於每個節點的值都大於等於子樹中每個節點值的堆，我們叫作“大頂堆”。對於每個節點的值都小於等於子樹中每個節點值的堆，我們叫作“小頂堆”。
堆實現
完全二叉樹比較適合用數組來存儲（鏈表也可以實現）。為什麽這麽說呢？用數組來存儲完全二叉樹是非常節省存儲空間的。因為我們不需要存儲左右子節點的指針，單純地通過數組的下標，就可以找到一個節點的左右子節點和父節點。
經過上圖可以發現，數組位置0為空，雖然浪費了一個存儲空間，但是當計算元素在數組位置的時候確非常方便：數組下標為X的元素的左子樹的下標為2x，右子樹的下標為2x+1。
其實實現一個堆非常簡單，就是順著元素所在的路徑，向上或者向下對比然後交換位置
1. 添加元素
添加元素的時候我們習慣采用自下而上的調整方式來調整堆，我們在數組的最後一個空閑位置插入新元素，按照堆的下標上標原則查找到父元素對比，如果小於父元素的值（大頂堆），則互相交換。如圖：
2. 刪除最大（最小元素）
對於大頂堆，堆頂的元素就是最大元素。刪除該元素之後，我們需要把第二大元素提到堆頂位置。依次類推，直到把路徑上的所有元素都調整完畢。
擴展閱讀
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();
}
結果如下:
突破內存限制的高性能排序