C#实现外排序的示例代码
作者:神仙别闹
一、N 路归并排序
1.1、概序
我们知道算法中有一种叫做分治思想,一个大问题我们可以采取分而治之,各个突破,当子问题解决了,大问题也就 KO 了,还有一点我们知道内排序的归并排序是采用二路归并的,因为分治后有 LogN 层,每层两路归并需要 N 的时候,最后复杂度为 NlogN,那么外排序我们可以将这个“二”扩大到 M,也就是将一个大文件分成 M 个小文件,每个小文件是有序的,然后对应在内存中我们开 M 个优先队列,每个队列从对应编号的文件中读取 TopN 条记录,然后我们从 M 路队列中各取一个数字进入中转站队列,并将该数字打上队列编号标记,当从中转站出来的最小数字就是我们最后要排序的数字之一,因为该数字打上了队列编号,所以方便我们通知对应的编号队列继续出数字进入中转站队列,可以看出中转站一直保存了 M 个记录,当中转站中的所有数字都出队完毕,则外排序结束。如果大家有点蒙的话,我再配合一张图,相信大家就会一目了然,这考验的是我们的架构能力。
图中这里有个 Batch 容器,这个容器我是基于性能考虑的,当 batch=n 时,我们定时刷新到文件中,保证内存有足够的空间。
1.2、构建
<1> 生成数据
这个基本没什么好说的,采用随机数生成 n 条记录。
<2> 切分数据
根据实际情况我们来决定到底要分成多少个小文件,并且小文件的数据必须是有序的,小文件的个数也对应这内存中有多少个优先队列。
<3> 加入队列
我们知道内存队列存放的只是小文件的 topN 条记录,当内存队列为空时,我们需要再次从小文件中读取下一批的 TopN 条数据,然后放入中转站继续进行比较。
<4> 测试
最后我们来测试一下:
数据量:short.MaxValue。
内存存放量:1200。
在这种场景下,我们决定每个文件放 1000 条,也就有 33 个小文件,也就有 33 个内存队列,每个队列取 Top100 条,Batch=500 时刷新
硬盘,中转站存放 332 个数字(因为入中转站时打上了队列标记),最后内存活动最大总数为:sum=33100+500+66=896<1200。
时间复杂度为 N*logN。
总的代码:
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 } }
到此这篇关于C#实现外排序的示例代码的文章就介绍到这了,更多相关C# 外排序内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!