外排序,当然在我看来,外排序考验的是一个程序员的架构能力,而不仅仅局限于排序这个层次。
我们知道算法中有一种叫做分治思想,一个大问题我们可以采取分而治之,各个突破,当子问题解决了,大问题也就KO了,还有一点我们知道
内排序的归并排序是采用二路归并的,因为分治后有LogN层,每层两路归并需要N的时候,最后复杂度为NlogN,那么外排序我们可以将这个“二”
扩大到M,也就是将一个大文件分成M个小文件,每个小文件是有序的,然后对应在内存中我们开M个优先队列,每个队列从对应编号的文件中读取
TopN条记录,然后我们从M路队列中各取一个数字进入中转站队列,并将该数字打上队列编号标记,当从中转站出来的最小数字就是我们最后要排
序的数字之一,因为该数字打上了队列编号,所以方便我们通知对应的编号队列继续出数字进入中转站队列,可以看出中转站一直保存了M个记录,
当中转站中的所有数字都出队完毕,则外排序结束。如果大家有点蒙的话,我再配合一张图,相信大家就会一目了然,这考验的是我们的架构能力。
图中这里有个Batch容器,这个容器我是基于性能考虑的,当batch=n时,我们定时刷新到文件中,保证内存有足够的空间。
这个基本没什么好说的,采用随机数生成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
根据实际情况我们来决定到底要分成多少个小文件,并且小文件的数据必须是有序的,小文件的个数也对应这内存中有多少个优先队列。
- #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
我们知道内存队列存放的只是小文件的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
最后我们来测试一下:
数据量: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();
- }
- using System;
- using System.Collections.Generic;
- using System.Linq;
- using System.Text;
- using System.Diagnostics;
- using System.Threading;
- using System.IO;
- using System.Threading.Tasks;
- namespace ConsoleApplication2
- {
- public class Program
- {
- 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();
- }
- #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
- #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
- #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
- }
- }
- using System;
- using System.Collections.Generic;
- using System.Linq;
- using System.Text;
- using System.Diagnostics;
- using System.Threading;
- using System.IO;
- namespace ConsoleApplication2
- {
- public class PriorityQueue<T>
- {
- /// <summary>
- /// 定义一个数组来存放节点
- /// </summary>
- private List<HeapNode> nodeList = new List<HeapNode>();
- #region 堆节点定义
- /// <summary>
- /// 堆节点定义
- /// </summary>
- public class HeapNode
- {
- /// <summary>
- /// 实体数据
- /// </summary>
- public T t { get; set; }
- /// <summary>
- /// 优先级别 1-10个级别 (优先级别递增)
- /// </summary>
- public int level { get; set; }
- public HeapNode(T t, int level)
- {
- this.t = t;
- this.level = level;
- }
- public HeapNode() { }
- }
- #endregion
- #region 添加操作
- /// <summary>
- /// 添加操作
- /// </summary>
- public void Eequeue(T t, int level = 1)
- {
- //将当前节点追加到堆尾
- nodeList.Add(new HeapNode(t, level));
- //如果只有一个节点,则不需要进行筛操作
- if (nodeList.Count == 1)
- return;
- //获取最后一个非叶子节点
- int parent = nodeList.Count / 2 - 1;
- //堆调整
- UpHeapAdjust(nodeList, parent);
- }
- #endregion
- #region 对堆进行上滤操作,使得满足堆性质
- /// <summary>
- /// 对堆进行上滤操作,使得满足堆性质
- /// </summary>
- /// <param name="nodeList"></param>
- /// <param name="index">非叶子节点的之后指针(这里要注意:我们
- /// 的筛操作时针对非叶节点的)
- /// </param>
- public void UpHeapAdjust(List<HeapNode> nodeList, int parent)
- {
- while (parent >= 0)
- {
- //当前index节点的左孩子
- var left = 2 * parent + 1;
- //当前index节点的右孩子
- var right = left + 1;
- //parent子节点中最大的孩子节点,方便于parent进行比较
- //默认为left节点
- var min = left;
- //判断当前节点是否有右孩子
- if (right < nodeList.Count)
- {
- //判断parent要比较的最大子节点
- min = nodeList[left].level < nodeList[right].level ? left : right;
- }
- //如果parent节点大于它的某个子节点的话,此时筛操作
- if (nodeList[parent].level > nodeList[min].level)
- {
- //子节点和父节点进行交换操作
- var temp = nodeList[parent];
- nodeList[parent] = nodeList[min];
- nodeList[min] = temp;
- //继续进行更上一层的过滤
- parent = (int)Math.Ceiling(parent / 2d) - 1;
- }
- else
- {
- break;
- }
- }
- }
- #endregion
- #region 优先队列的出队操作
- /// <summary>
- /// 优先队列的出队操作
- /// </summary>
- /// <returns></returns>
- public HeapNode Dequeue()
- {
- if (nodeList.Count == 0)
- return null;
- //出队列操作,弹出数据头元素
- var pop = nodeList[0];
- //用尾元素填充头元素
- nodeList[0] = nodeList[nodeList.Count - 1];
- //删除尾节点
- nodeList.RemoveAt(nodeList.Count - 1);
- //然后从根节点下滤堆
- DownHeapAdjust(nodeList, 0);
- return pop;
- }
- #endregion
- #region 对堆进行下滤操作,使得满足堆性质
- /// <summary>
- /// 对堆进行下滤操作,使得满足堆性质
- /// </summary>
- /// <param name="nodeList"></param>
- /// <param name="index">非叶子节点的之后指针(这里要注意:我们
- /// 的筛操作时针对非叶节点的)
- /// </param>
- public void DownHeapAdjust(List<HeapNode> nodeList, int parent)
- {
- while (2 * parent + 1 < nodeList.Count)
- {
- //当前index节点的左孩子
- var left = 2 * parent + 1;
- //当前index节点的右孩子
- var right = left + 1;
- //parent子节点中最大的孩子节点,方便于parent进行比较
- //默认为left节点
- var min = left;
- //判断当前节点是否有右孩子
- if (right < nodeList.Count)
- {
- //判断parent要比较的最大子节点
- min = nodeList[left].level < nodeList[right].level ? left : right;
- }
- //如果parent节点小于它的某个子节点的话,此时筛操作
- if (nodeList[parent].level > nodeList[min].level)
- {
- //子节点和父节点进行交换操作
- var temp = nodeList[parent];
- nodeList[parent] = nodeList[min];
- nodeList[min] = temp;
- //继续进行更下一层的过滤
- parent = min;
- }
- else
- {
- break;
- }
- }
- }
- #endregion
- #region 获取元素并下降到指定的level级别
- /// <summary>
- /// 获取元素并下降到指定的level级别
- /// </summary>
- /// <returns></returns>
- public HeapNode GetAndDownPriority(int level)
- {
- if (nodeList.Count == 0)
- return null;
- //获取头元素
- var pop = nodeList[0];
- //设置指定优先级(如果为 MinValue 则为 -- 操作)
- nodeList[0].level = level == int.MinValue ? --nodeList[0].level : level;
- //下滤堆
- DownHeapAdjust(nodeList, 0);
- return nodeList[0];
- }
- #endregion
- #region 获取元素并下降优先级
- /// <summary>
- /// 获取元素并下降优先级
- /// </summary>
- /// <returns></returns>
- public HeapNode GetAndDownPriority()
- {
- //下降一个优先级
- return GetAndDownPriority(int.MinValue);
- }
- #endregion
- #region 返回当前优先队列中的元素个数
- /// <summary>
- /// 返回当前优先队列中的元素个数
- /// </summary>
- /// <returns></returns>
- public int Count()
- {
- return nodeList.Count;
- }
- #endregion
- }
- }