第十八题 外排序

外排序，当然在我看来，外排序考验的是一个程序员的架构能力，而不仅仅局限于排序这个层次。

一：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();
      }

总的代码：

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
    }
}

• 上一篇：第十七题 Dijkstra算法
• 第十九题 双端队列：下一篇