說到排序，大家第一反應基本上是內排序，是的，算法嘛，玩的就是內存，然而內存是有限制的，總有裝不下的那一天，此時就可以來玩玩
外排序，當然在我看來，外排序考驗的是一個程序員的架構能力，而不僅僅局限於排序這個層次。
一：N路歸並排序
1.概序
我們知道算法中有一種叫做分治思想，一個大問題我們可以采取分而治之，各個突破，當子問題解決了，大問題也就KO了，還有一點我們知道
內排序的歸並排序是采用二路歸並的，因為分治后有LogN層，每層兩路歸並需要N的時候，最后復雜度為NlogN，那么外排序我們可以將這個“二”
擴大到M，也就是將一個大文件分成M個小文件，每個小文件是有序的，然后對應在內存中我們開M個優先隊列，每個隊列從對應編號的文件中讀取
TopN條記錄，然后我們從M路隊列中各取一個數字進入中轉站隊列，並將該數字打上隊列編號標記，當從中轉站出來的最小數字就是我們最后要排
序的數字之一，因為該數字打上了隊列編號，所以方便我們通知對應的編號隊列繼續出數字進入中轉站隊列，可以看出中轉站一直保存了M個記錄，
當中轉站中的所有數字都出隊完畢，則外排序結束。如果大家有點蒙的話，我再配合一張圖，相信大家就會一目了然，這考驗的是我們的架構能力。
圖中這里有個Batch容器，這個容器我是基於性能考慮的，當batch=n時，我們定時刷新到文件中，保證內存有足夠的空間。
2.構建
<1> 生成數據
這個基本沒什么好說的，采用隨機數生成n條記錄。
#region 隨機生成數據
/// <summary>
/// 隨機生成數據
///<param name="max">執行生成的數據上線</param>
/// </summary>
public static void CreateData(int max)
{
var sw = new StreamWriter(Environment.CurrentDirectory + "//demo.txt");
for (int i = 0; i < max; i++)
{
Thread.Sleep(2);
var rand = new Random((int)DateTime.Now.Ticks).Next(0, int.MaxValue >> 3);
sw.WriteLine(rand);
}
sw.Close();
}
#endregion
<2> 切分數據
根據實際情況我們來決定到底要分成多少個小文件，並且小文件的數據必須是有序的，小文件的個數也對應這內存中有多少個優先隊列。
#region 將數據進行分份
/// <summary>
/// 將數據進行分份
/// <param name="size">每頁要顯示的條數</param>
/// </summary>
public static int Split(int size)
{
//文件總記錄數
int totalCount = 0;
//每一份文件存放 size 條 記錄
List<int> small = new List<int>();
var sr = new StreamReader((Environment.CurrentDirectory + "//demo.txt"));
var pageSize = size;
int pageCount = 0;
int pageIndex = 0;
while (true)
{
var line = sr.ReadLine();
if (!string.IsNullOrEmpty(line))
{
totalCount++;
//加入小集合中
small.Add(Convert.ToInt32(line));
//說明已經到達指定的 size 條數了
if (totalCount % pageSize == 0)
{
pageIndex = totalCount / pageSize;
small = small.OrderBy(i => i).Select(i => i).ToList();
File.WriteAllLines(Environment.CurrentDirectory + "//" + pageIndex + ".txt", small.Select(i => i.ToString()));
small.Clear();
}
}
else
{
//說明已經讀完了，將剩余的small記錄寫入到文件中
pageCount = (int)Math.Ceiling((double)totalCount / pageSize);
small = small.OrderBy(i => i).Select(i => i).ToList();
File.WriteAllLines(Environment.CurrentDirectory + "//" + pageCount + ".txt", small.Select(i => i.ToString()));
break;
}
}
return pageCount;
}
#endregion
<3> 加入隊列
我們知道內存隊列存放的只是小文件的topN條記錄，當內存隊列為空時，我們需要再次從小文件中讀取下一批的TopN條數據，然后放入中轉站
繼續進行比較。
#region 將數據加入指定編號隊列
/// <summary>
/// 將數據加入指定編號隊列
/// </summary>
/// <param name="i">隊列編號</param>
/// <param name="skip">文件中跳過的條數</param>
/// <param name="list"></param>
/// <param name="top">需要每次讀取的條數</param>
public static void AddQueue(int i, List<PriorityQueue<int?>> list, ref int[] skip, int top = 100)
{
var result = File.ReadAllLines((Environment.CurrentDirectory + "//" + (i + 1) + ".txt"))
.Skip(skip[i]).Take(top).Select(j => Convert.ToInt32(j));
//加入到集合中
foreach (var item in result)
list[i].Eequeue(null, item);
//將個數累計到skip中，表示下次要跳過的記錄數
skip[i] += result.Count();
}
#endregion
<4> 測試
最后我們來測試一下:
數據量：short.MaxValue。
內存存放量：1200。
在這種場景下，我們決定每個文件放1000條，也就有33個小文件，也就有33個內存隊列，每個隊列取Top100條，Batch=500時刷新
硬盤，中轉站存放33*2個數字（因為入中轉站時打上了隊列標記）,最后內存活動最大總數為：sum=33*100+500+66=896<1200。
時間復雜度為N*logN。當然這個“閥值”，我們可以再仔細微調。
public static void Main()
{
//生成2^15數據
CreateData(short.MaxValue);
//每個文件存放1000條
var pageSize = 1000;
//達到batchCount就刷新記錄
var batchCount = 0;
//判斷需要開啟的隊列
var pageCount = Split(pageSize);
//內存限制：1500條
List<PriorityQueue<int?>> list = new List<PriorityQueue<int?>>();
//定義一個隊列中轉器
PriorityQueue<int?> queueControl = new PriorityQueue<int?>();
//定義每個隊列完成狀態
bool[] complete = new bool[pageCount];
//隊列讀取文件時應該跳過的記錄數
int[] skip = new int[pageCount];
//是否所有都完成了
int allcomplete = 0;
//定義 10 個隊列
for (int i = 0; i < pageCount; i++)
{
list.Add(new PriorityQueue<int?>());
//i： 記錄當前的隊列編碼
//list: 隊列數據
//skip：跳過的條數
AddQueue(i, list, ref skip);
}
//初始化操作，從每個隊列中取出一條記錄，並且在入隊的過程中
//記錄該數據所屬的 “隊列編號”
for (int i = 0; i < list.Count; i++)
{
var temp = list[i].Dequeue();
//i:隊列編碼,level:要排序的數據
queueControl.Eequeue(i, temp.level);
}
//默認500條寫入一次文件
List<int> batch = new List<int>();
//記錄下次應該從哪一個隊列中提取數據
int nextIndex = 0;
while (queueControl.Count() > 0)
{
//從中轉器中提取數據
var single = queueControl.Dequeue();
//記錄下一個隊列總應該出隊的數據
nextIndex = single.t.Value;
var nextData = list[nextIndex].Dequeue();
//如果改對內彈出為null，則說明該隊列已經，需要從nextIndex文件中讀取數據
if (nextData == null)
{
//如果該隊列沒有全部讀取完畢
if (!complete[nextIndex])
{
AddQueue(nextIndex, list, ref skip);
//如果從文件中讀取還是沒有，則說明改文件中已經沒有數據了
if (list[nextIndex].Count() == 0)
{
complete[nextIndex] = true;
allcomplete++;
}
else
{
nextData = list[nextIndex].Dequeue();
}
}
}
//如果彈出的數不為空，則將該數入中轉站
if (nextData != null)
{
//將要出隊的數據 轉入 中轉站
queueControl.Eequeue(nextIndex, nextData.level);
}
batch.Add(single.level);
//如果batch=500，或者所有的文件都已經讀取完畢，此時我們要批量刷入數據
if (batch.Count == batchCount || allcomplete == pageCount)
{
var sw = new StreamWriter(Environment.CurrentDirectory + "//result.txt", true);
foreach (var item in batch)
{
sw.WriteLine(item);
}
sw.Close();
batch.Clear();
}
}
Console.WriteLine("恭喜，外排序完畢！");
Console.Read();
}
總的代碼：
View Code
View Code
1 using System;
2 using System.Collections.Generic;
3 using System.Linq;
4 using System.Text;
5 using System.Diagnostics;
6 using System.Threading;
7 using System.IO;
8 using System.Threading.Tasks;
9
10 namespace ConsoleApplication2
11 {
12 public class Program
13 {
14 public static void Main()
15 {
16 //生成2^15數據
17 CreateData(short.MaxValue);
18
19 //每個文件存放1000條
20 var pageSize = 1000;
21
22 //達到batchCount就刷新記錄
23 var batchCount = 0;
24
25 //判斷需要開啟的隊列
26 var pageCount = Split(pageSize);
27
28 //內存限制：1500條
29 List<PriorityQueue<int?>> list = new List<PriorityQueue<int?>>();
30
31 //定義一個隊列中轉器
32 PriorityQueue<int?> queueControl = new PriorityQueue<int?>();
33
34 //定義每個隊列完成狀態
35 bool[] complete = new bool[pageCount];
36
37 //隊列讀取文件時應該跳過的記錄數
38 int[] skip = new int[pageCount];
39
40 //是否所有都完成了
41 int allcomplete = 0;
42
43 //定義 10 個隊列
44 for (int i = 0; i < pageCount; i++)
45 {
46 list.Add(new PriorityQueue<int?>());
47
48 //i： 記錄當前的隊列編碼
49 //list: 隊列數據
50 //skip：跳過的條數
51 AddQueue(i, list, ref skip);
52 }
53
54 //初始化操作，從每個隊列中取出一條記錄，並且在入隊的過程中
55 //記錄該數據所屬的 “隊列編號”
56 for (int i = 0; i < list.Count; i++)
57 {
58 var temp = list[i].Dequeue();
59
60 //i:隊列編碼,level:要排序的數據
61 queueControl.Eequeue(i, temp.level);
62 }
63
64 //默認500條寫入一次文件
65 List<int> batch = new List<int>();
66
67 //記錄下次應該從哪一個隊列中提取數據
68 int nextIndex = 0;
69
70 while (queueControl.Count() > 0)
71 {
72 //從中轉器中提取數據
73 var single = queueControl.Dequeue();
74
75 //記錄下一個隊列總應該出隊的數據
76 nextIndex = single.t.Value;
77
78 var nextData = list[nextIndex].Dequeue();
79
80 //如果改對內彈出為null，則說明該隊列已經，需要從nextIndex文件中讀取數據
81 if (nextData == null)
82 {
83 //如果該隊列沒有全部讀取完畢
84 if (!complete[nextIndex])
85 {
86 AddQueue(nextIndex, list, ref skip);
87
88 //如果從文件中讀取還是沒有，則說明改文件中已經沒有數據了
89 if (list[nextIndex].Count() == 0)
90 {
91 complete[nextIndex] = true;
92 allcomplete++;
93 }
94 else
95 {
96 nextData = list[nextIndex].Dequeue();
97 }
98 }
99 }
100
101 //如果彈出的數不為空，則將該數入中轉站
102 if (nextData != null)
103 {
104 //將要出隊的數據 轉入 中轉站
105 queueControl.Eequeue(nextIndex, nextData.level);
106 }
107
108 batch.Add(single.level);
109
110 //如果batch=500，或者所有的文件都已經讀取完畢，此時我們要批量刷入數據
111 if (batch.Count == batchCount || allcomplete == pageCount)
112 {
113 var sw = new StreamWriter(Environment.CurrentDirectory + "//result.txt", true);
114
115 foreach (var item in batch)
116 {
117 sw.WriteLine(item);
118 }
119
120 sw.Close();
121
122 batch.Clear();
123 }
124 }
125
126 Console.WriteLine("恭喜，外排序完畢！");
127 Console.Read();
128 }
129
130 #region 將數據加入指定編號隊列
131 /// <summary>
132 /// 將數據加入指定編號隊列
133 /// </summary>
134 /// <param name="i">隊列編號</param>
135 /// <param name="skip">文件中跳過的條數</param>
136 /// <param name="list"></param>
137 /// <param name="top">需要每次讀取的條數</param>
138 public static void AddQueue(int i, List<PriorityQueue<int?>> list, ref int[] skip, int top = 100)
139 {
140 var result = File.ReadAllLines((Environment.CurrentDirectory + "//" + (i + 1) + ".txt"))
141 .Skip(skip[i]).Take(top).Select(j => Convert.ToInt32(j));
142
143 //加入到集合中
144 foreach (var item in result)
145 list[i].Eequeue(null, item);
146
147 //將個數累計到skip中，表示下次要跳過的記錄數
148 skip[i] += result.Count();
149 }
150 #endregion
151
152 #region 隨機生成數據
153 /// <summary>
154 /// 隨機生成數據
155 ///<param name="max">執行生成的數據上線</param>
156 /// </summary>
157 public static void CreateData(int max)
158 {
159 var sw = new StreamWriter(Environment.CurrentDirectory + "//demo.txt");
160
161 for (int i = 0; i < max; i++)
162 {
163 Thread.Sleep(2);
164 var rand = new Random((int)DateTime.Now.Ticks).Next(0, int.MaxValue >> 3);
165
166 sw.WriteLine(rand);
167 }
168 sw.Close();
169 }
170 #endregion
171
172 #region 將數據進行分份
173 /// <summary>
174 /// 將數據進行分份
175 /// <param name="size">每頁要顯示的條數</param>
176 /// </summary>
177 public static int Split(int size)
178 {
179 //文件總記錄數
180 int totalCount = 0;
181
182 //每一份文件存放 size 條 記錄
183 List<int> small = new List<int>();
184
185 var sr = new StreamReader((Environment.CurrentDirectory + "//demo.txt"));
186
187 var pageSize = size;
188
189 int pageCount = 0;
190
191 int pageIndex = 0;
192
193 while (true)
194 {
195 var line = sr.ReadLine();
196
197 if (!string.IsNullOrEmpty(line))
198 {
199 totalCount++;
200
201 //加入小集合中
202 small.Add(Convert.ToInt32(line));
203
204 //說明已經到達指定的 size 條數了
205 if (totalCount % pageSize == 0)
206 {
207 pageIndex = totalCount / pageSize;
208
209 small = small.OrderBy(i => i).Select(i => i).ToList();
210
211 File.WriteAllLines(Environment.CurrentDirectory + "//" + pageIndex + ".txt", small.Select(i => i.ToString()));
212
213 small.Clear();
214 }
215 }
216 else
217 {
218 //說明已經讀完了，將剩余的small記錄寫入到文件中
219 pageCount = (int)Math.Ceiling((double)totalCount / pageSize);
220
221 small = small.OrderBy(i => i).Select(i => i).ToList();
222
223 File.WriteAllLines(Environment.CurrentDirectory + "//" + pageCount + ".txt", small.Select(i => i.ToString()));
224
225 break;
226 }
227 }
228
229 return pageCount;
230 }
231 #endregion
232 }
233 }
優先隊列：
View Code
1 using System;
2 using System.Collections.Generic;
3 using System.Linq;
4 using System.Text;
5 using System.Diagnostics;
6 using System.Threading;
7 using System.IO;
8
9 namespace ConsoleApplication2
10 {
11 public class PriorityQueue<T>
12 {
13 /// <summary>
14 /// 定義一個數組來存放節點
15 /// </summary>
16 private List<HeapNode> nodeList = new List<HeapNode>();
17
18 #region 堆節點定義
19 /// <summary>
20 /// 堆節點定義
21 /// </summary>
22 public class HeapNode
23 {
24 /// <summary>
25 /// 實體數據
26 /// </summary>
27 public T t { get; set; }
28
29 /// <summary>
30 /// 優先級別 1-10個級別 (優先級別遞增)
31 /// </summary>
32 public int level { get; set; }
33
34 public HeapNode(T t, int level)
35 {
36 this.t = t;
37 this.level = level;
38 }
39
40 public HeapNode() { }
41 }
42 #endregion
43
44 #region 添加操作
45 /// <summary>
46 /// 添加操作
47 /// </summary>
48 public void Eequeue(T t, int level = 1)
49 {
50 //將當前節點追加到堆尾
51 nodeList.Add(new HeapNode(t, level));
52
53 //如果只有一個節點，則不需要進行篩操作
54 if (nodeList.Count == 1)
55 return;
56
57 //獲取最后一個非葉子節點
58 int parent = nodeList.Count / 2 - 1;
59
60 //堆調整
61 UpHeapAdjust(nodeList, parent);
62 }
63 #endregion
64
65 #region 對堆進行上濾操作，使得滿足堆性質
66 /// <summary>
67 /// 對堆進行上濾操作，使得滿足堆性質
68 /// </summary>
69 /// <param name="nodeList"></param>
70 /// <param name="index">非葉子節點的之后指針（這里要注意：我們
71 /// 的篩操作時針對非葉節點的）
72 /// </param>
73 public void UpHeapAdjust(List<HeapNode> nodeList, int parent)
74 {
75 while (parent >= 0)
76 {
77 //當前index節點的左孩子
78 var left = 2 * parent + 1;
79
80 //當前index節點的右孩子
81 var right = left + 1;
82
83 //parent子節點中最大的孩子節點，方便於parent進行比較
84 //默認為left節點
85 var min = left;
86
87 //判斷當前節點是否有右孩子
88 if (right < nodeList.Count)
89 {
90 //判斷parent要比較的最大子節點
91 min = nodeList[left].level < nodeList[right].level ? left : right;
92 }
93
94 //如果parent節點大於它的某個子節點的話，此時篩操作
95 if (nodeList[parent].level > nodeList[min].level)
96 {
97 //子節點和父節點進行交換操作
98 var temp = nodeList[parent];
99 nodeList[parent] = nodeList[min];
100 nodeList[min] = temp;
101
102 //繼續進行更上一層的過濾
103 parent = (int)Math.Ceiling(parent / 2d) - 1;
104 }
105 else
106 {
107 break;
108 }
109 }
110 }
111 #endregion
112
113 #region 優先隊列的出隊操作
114 /// <summary>
115 /// 優先隊列的出隊操作
116 /// </summary>
117 /// <returns></returns>
118 public HeapNode Dequeue()
119 {
120 if (nodeList.Count == 0)
121 return null;
122
123 //出隊列操作，彈出數據頭元素
124 var pop = nodeList[0];
125
126 //用尾元素填充頭元素
127 nodeList[0] = nodeList[nodeList.Count - 1];
128
129 //刪除尾節點
130 nodeList.RemoveAt(nodeList.Count - 1);
131
132 //然后從根節點下濾堆
133 DownHeapAdjust(nodeList, 0);
134
135 return pop;
136 }
137 #endregion
138
139 #region 對堆進行下濾操作，使得滿足堆性質
140 /// <summary>
141 /// 對堆進行下濾操作，使得滿足堆性質
142 /// </summary>
143 /// <param name="nodeList"></param>
144 /// <param name="index">非葉子節點的之后指針（這里要注意：我們
145 /// 的篩操作時針對非葉節點的）
146 /// </param>
147 public void DownHeapAdjust(List<HeapNode> nodeList, int parent)
148 {
149 while (2 * parent + 1 < nodeList.Count)
150 {
151 //當前index節點的左孩子
152 var left = 2 * parent + 1;
153
154 //當前index節點的右孩子
155 var right = left + 1;
156
157 //parent子節點中最大的孩子節點，方便於parent進行比較
158 //默認為left節點
159 var min = left;
160
161 //判斷當前節點是否有右孩子
162 if (right < nodeList.Count)
163 {
164 //判斷parent要比較的最大子節點
165 min = nodeList[left].level < nodeList[right].level ? left : right;
166 }
167
168 //如果parent節點小於它的某個子節點的話，此時篩操作
169 if (nodeList[parent].level > nodeList[min].level)
170 {
171 //子節點和父節點進行交換操作
172 var temp = nodeList[parent];
173 nodeList[parent] = nodeList[min];
174 nodeList[min] = temp;
175
176 //繼續進行更下一層的過濾
177 parent = min;
178 }
179 else
180 {
181 break;
182 }
183 }
184 }
185 #endregion
186
187 #region 獲取元素並下降到指定的level級別
188 /// <summary>
189 /// 獲取元素並下降到指定的level級別
190 /// </summary>
191 /// <returns></returns>
192 public HeapNode GetAndDownPriority(int level)
193 {
194 if (nodeList.Count == 0)
195 return null;
196
197 //獲取頭元素
198 var pop = nodeList[0];
199
200 //設置指定優先級（如果為 MinValue 則為 -- 操作）
201 nodeList[0].level = level == int.MinValue ? --nodeList[0].level : level;
202
203 //下濾堆
204 DownHeapAdjust(nodeList, 0);
205
206 return nodeList[0];
207 }
208 #endregion
209
210 #region 獲取元素並下降優先級
211 /// <summary>
212 /// 獲取元素並下降優先級
213 /// </summary>
214 /// <returns></returns>
215 public HeapNode GetAndDownPriority()
216 {
217 //下降一個優先級
218 return GetAndDownPriority(int.MinValue);
219 }
220 #endregion
221
222 #region 返回當前優先隊列中的元素個數
223 /// <summary>
224 /// 返回當前優先隊列中的元素個數
225 /// </summary>
226 /// <returns></returns>
227 public int Count()
228 {
229 return nodeList.Count;
230 }
231 #endregion
232 }
233 }