程序員如何高性能排序多個文件？

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

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

問題分析

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

• 如何能把所有的日誌文件按照時間排序？
• 日誌文件的總大小為500M*100 ，大約50G，所以全部載入到內存是不可能的；
• 程序執行過程中，要頻繁排序並查找最小元素。

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

解決方案

堆定義

堆（heap）是計算機科學中一類特殊的數據結構的統稱，通常是一個可以被看做一棵樹的數組對象。

堆總是滿足下列性質：

• 堆中某個節點的值總是不大於或不小於其父節點的值；
• 堆總是一棵完全二叉樹（完全二叉樹要求，除了最後一層，其他層的節點個數都是滿的，最後一層的節點都靠左排列）。

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

堆實現

完全二叉樹比較適合用數組來存儲（鏈表也可以實現）。為什麼這麼說呢？用數組來存儲完全二叉樹是非常節省存儲空間的。因為我們不需要存儲左右子節點的指針，單純地通過數組的下標，就可以找到一個節點的左右子節點和父節點。

經過上圖可以發現，數組位置0為空，雖然浪費了一個存儲空間，但是當計算元素在數組位置的時候確非常方便：數組下標為X的元素的左子樹的下標為2x，右子樹的下標為2x+1。

其實實現一個堆非常簡單，就是順著元素所在的路徑，向上或者向下對比然後交換位置。

1、添加元素

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

2、刪除最大（最小元素）

對於大頂堆，堆頂的元素就是最大元素。刪除該元素之後，我們需要把第二大元素提到堆頂位置。依次類推，直到把路徑上的所有元素都調整完畢。

總結

• 小頂堆的頂部元素其實就是整個堆最小的元素，大頂堆頂部元素是整個堆的最大元素。這也是堆排序的最大優點，取最小元素或者最大元素時間複雜度為O(1)。
• 刪除元素的時候我們要注意一點，如果採用自頂向下交換元素的方式，在很多情況下造成堆嚴重的不平衡（左右子樹深度相差較大）的情況，為了防止類似情況，我們可以把最後一個元素提到堆頂，然後調整的策略，因為最後一個元素總是在最後一級，不會造成左右子樹相差很大的情況。
• 對於有重複元素的堆，一種解決方法是認為是誰先誰大，後進入堆的元素小於先進入堆的元素，這樣在查找的時候一定要查徹底才行。另外一種方式是在堆的每個元素中存儲一個鏈表，用來存放相同的元素，原理類似於散列表。不過這樣在刪除這個元素的時候需要特殊處理一下。

• 刪除堆頂數據和往堆中插入數據的時間複雜度都是O(logn)。
• 不斷調整堆的過程其實就是排序過程，在某些場景下，我們可以利用堆來實現排序。

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)+ "
");
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 otal.txt", JsonConvert.SerializeObject(heapFirstItem.Data) + "
");
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();
}

最終排序結果如下圖：

機器使用CPU內存完全沒有達到所有排序文件的總大小：

作者：菜菜，一個奔走在通往互聯網更高之路的工程師，熱衷於互聯網技術。目前就職於某互聯網教育公司，應用服務端主要負責人。擁有10年+互聯網開發經驗。熱衷於高性能、高並發、分散式技術領域的研究。 主要工作語言為C#和Golang 。

聲明：本文為作者投稿，版權歸對方所有，編輯郭芮。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！