说到排序，大家第一反应基本上是内排序，是的，演算法嘛，玩的就是内存，然而内存是有限制的，总有装不下的那一天，此时就可以来玩玩
外排序，当然在我看来，外排序考验的是一个程式设计师的架构能力，而不仅仅局限于排序这个层次。
一：N 路归并排序
1. 概序
我们知道演算法中有一种叫做分治思想，一个大问题我们可以采取分而治之，各个突破，当子问题解决了，大问题也就 KO 了，还有一点我们知道
内排序的归并排序是采用二路归并的，因为分治后有 LogN 层，每层两路归并需要 N 的时候，最后复杂度为 NlogN，那么外排序我们可以将这个 “二”
扩大到 M，也就是将一个大档案分成 M 个小档案，每个小档案是有序的，然后对应在内存中我们开 M 个优先伫列，每个伫列从对应编号的档案中读取
TopN 条记录，然后我们从 M 路伫列中各取一个数字进入中转站伫列，并将该数字打上伫列编号标记，当从中转站出来的最小数字就是我们最后要排
序的数字之一，因为该数字打上了伫列编号，所以方便我们通知对应的编号伫列继续出数字进入中转站伫列，可以看出中转站一直储存了 M 个记录，
当中转站中的所有数字都出队完毕，则外排序结束。如果大家有点蒙的话，我再配合一张图，相信大家就会一目了然，这考验的是我们的架构能力。
图中这里有个 Batch 容器，这个容器我是基于效能考虑的，当 batch=n 时，我们定时重新整理到档案中，保证内存有足够的空间。
2. 构建
<1> 生成资料
这个基本没什么好说的，采用随机数生成 n 条记录。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#region 随机生成资料
///
///执行生成的资料上线 ///
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> 切分资料
根据实际情况我们来决定到底要分成多少个小档案，并且小档案的资料必须是有序的，小档案的个数也对应这内存中有多少个优先伫列。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
#region 将资料进行分份
///
/// 每页要显示的条数 ///
public static int Split(int size)
{
//档案总记录数
int totalCount = 0;
//每一份档案存放 size 条 记录
List
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 条资料，然后放入中转站
继续进行比较。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#region 将资料加入指定编号伫列
///
///
/// 伫列编号
/// 档案中跳过的条数
///
/// 需要每次读取的条数
public static void AddQueue(int i, List
{
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 。当然这个 “阀值”，我们可以再仔细微调。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
public static void Main()
{
//生成 2^15 资料
CreateData(short.MaxValue);
//每个档案存放 1000 条
var pageSize = 1000;
//达到 batchCount 就重新整理记录
var batchCount = 0;
//判断需要开启的伫列
var pageCount = Split(pageSize);
//内存限制：1500 条
List
//定义一个伫列中转器
PriorityQueue
//定义每个伫列完成状态
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
//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 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
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
30
31 //定义一个伫列中转器
32 PriorityQueue
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
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
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 ///
133 ///
134 /// 伫列编号
135 /// 档案中跳过的条数
136 ///
137 /// 需要每次读取的条数
138 public static void AddQueue(int i, List
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 ///
155 ///执行生成的资料上线 156 ///
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 ///
175 /// 每页要显示的条数 176 ///
177 public static int Split(int size)
178 {
179 //档案总记录数
180 int totalCount = 0;
181
182 //每一份档案存放 size 条 记录
183 List
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
12 {
13 ///
15 ///
16 private List
17
18 #region 堆节点定义
19 ///
21 ///
22 public class HeapNode
23 {
24 ///
26 ///
27 public T t { get; set; }
28
29 ///
31 ///
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 ///
47 ///
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 ///
68 ///
69 ///
70 /// 非叶子节点的之后指标（这里要注意：我们
71 /// 的筛操作时针对非叶节点的）
72 ///
73 public void UpHeapAdjust(List
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 ///
116 ///
117 ///
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 ///
142 ///
143 ///
144 /// 非叶子节点的之后指标（这里要注意：我们
145 /// 的筛操作时针对非叶节点的）
146 ///
147 public void DownHeapAdjust(List
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 ///
190 ///
191 ///
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 ///
213 ///
214 ///
215 public HeapNode GetAndDownPriority()
216 {
217 //下降一个优先顺序
218 return GetAndDownPriority(int.MinValue);
219 }
220 #endregion
221
222 #region 返回当前优先伫列中的元素个数
223 ///
225 ///
226 ///
227 public int Count()
228 {
229 return nodeList.Count;
230 }
231 #endregion
232 }
233 }
转载在互联网博客网站，版权归作者所有, 原文连结：http://www.cnblogs.com/huangxincheng/archive/2012/12/19/2824943.html