经度算法：磁盘多路归并排序

今天药忘吃喽~ 2022-03-16 08:44 183阅读 0赞

如果说语言的基础语法和业务逻辑编码的经验积累是术，那么数据结构与算法思想、设计模式就是道。就好像笑傲江湖里面华山派的剑宗、气宗一样，在最前期的时候剑宗的门人一般要比气宗的门人厉害，因为他们剑法精炼，但是到了后期，当其中的门人把内你练上去之后，他们则会比剑宗的人更加厉害。当然只偏向于剑法而忽略内力修为，或者只注重内力修为而不注重剑法修为都是不对的，我们应该两者并重。

多路归并排序在大数据领域也是常用的算法，常用于海量数据排序。

**什么时候需要流式多路归并排序？**

当数据量特别大时，这些数据无法被单个机器内存容纳，它需要被切分位多个集合分别由不同的机器进行内存排序（map 过程），然后再进行多路归并算法将来自多个不同机器的数据进行排序（reduce 过程）。这是我们平时讲流式（数据源来源于网络套接字）亦是如此。

![20190322222937942.png][]

**这样做有什么好处呢？**

多路归并排序的优势在于**内存消耗极低**，它的内存占用和输入文件的数量成正比，和**数据总量无关**，数据总量只会线性正比影响排序的时间。

**有实现思路？**

我们需要在内存里维护一个有序数组。每个输入文件当前最小的元素作为一个元素放在数组里。数组按照元素的大小保持排序状态。注：多个输入源（input-a、input-b...），每个输入源亦是从小到大排序好的

![20190322223658758.png][]

接下来就开始进入循环，循环的逻辑总是从最小的元素（左1）下手，在其所在的文件取出下一个元素，和当前数组中的元素进行比较。

1. 如果取出来的元素和当前数组中的最小元素相等，那么就可以直接将这个元素输出。再继续下一轮循环。不可能取出比当前数组最小元素还要小的元素，因为输入文件本身也是有序的。

![20190322224438403.png][]

2. 否则就需要将元素插入到当前的数组中的指定位置，继续保持数组有序。然后将数组中当前最小的元素输出并移除。再进行下一轮循环。

![20190322224545153.png][]

3. 如果遇到文件结尾，那就无法继续调用 next() 方法了，这时可以直接将数组中的最小元素输出并移除，数组也跟着变小了。再进行下一轮循环。当数组空了，说明所有的文件都处理完了，算法就可以结束了。

![20190322224726411.png][]

注：遍历替换过程中数组中永远不会存在同一个文件的两个元素

**如何查找一个数字在数组中要插入的位置？**

Java 内置了二分查找算法在使用上比较精巧，如果 key 可以在 list 中找到，那就直接返回相应的位置。如果找不到，它会返回负数，这个负数指明了插入的位置，也就是说在这个位置插入 key，数组将可以继续保持有序。

public class Collections {
      public static <T> int binarySearch(List<T> list, T key) {
        ...
        if (found) {
          return index;
        } else {
          return -(insertIndex+1);
        }
      }
    }

比如 binarySearch 返回了 index=-1，那么 insertIndex 就是 -(index+1)，也就是 0，插入点在数组开头。如果返回了 index=-size-1，那么 insertIndex 就是 size，是数组末尾。其它负数会插入数组中间。

![20190322225436210.png][]

**去实现它吧！**

public class DiskMergeSort implements Closeable {
    	private List<MergeSource> sources;
    	private MergeOut out;
    	public DiskMergeSort(List<String> files, String outFilename) {
    		this.sources = new ArrayList<>();
    		for (String filename : files) {
    			this.sources.add(new MergeSource(filename));
    		}
    		this.out = new MergeOut(outFilename);
    	}
    	//输出文件类
    	static class MergeOut implements Closeable {
    		private PrintWriter writer;
    		public MergeOut(String filename) {
    			try {
    				this.writer = new PrintWriter(new FileOutputStream(filename));
    			} catch (FileNotFoundException e) {
    			}
    		}
    		public void write(Bin bin) {
    			writer.println(bin.num);
    		}
    		@Override
    		public void close() throws IOException {
    			writer.flush();
    			writer.close();
    		}
    	}
    	//输入源：输入文件是有序的
    	static class MergeSource implements Closeable {
    		private BufferedReader reader;
    		private String cachedLine;
    
    		public MergeSource(String filename) {
    			try {
    				FileReader fr = new FileReader(filename);
    				this.reader = new BufferedReader(fr);
    			} catch (FileNotFoundException e) {
    			}
    		}
    		public boolean hasNext() {
    			String line;
    			try {
    				line = this.reader.readLine();
    				if (line == null || line.isEmpty()) {
    					return false;
    				}
    				this.cachedLine = line.trim();
    				return true;
    			} catch (IOException e) {
    			}
    			return false;
    		}
    		public int next() {
    			if (this.cachedLine == null) {
    				if (!hasNext()) {
    					throw new IllegalStateException("no content");
    				}
    			}
    			int num = Integer.parseInt(this.cachedLine);
    			this.cachedLine = null;
    			return num;
    		}
    		@Override
    		public void close() throws IOException {
    			this.reader.close();
    		}
    	}
    	//内存有序数组元素类:排序
    	static class Bin implements Comparable<Bin> {
    		int num;
    		MergeSource source;
    		Bin(MergeSource source, int num) {
    			this.source = source;
    			this.num = num;
    		}
    		@Override
    		public int compareTo(Bin o) {
    			return this.num - o.num;
    		}
    	}
    	//将每个输入文件的最小元素放入数组
    	public List<Bin> prepare() {
    		List<Bin> bins = new ArrayList<>();
    		for (MergeSource source : sources) {
    			Bin newBin = new Bin(source, source.next());
    			bins.add(newBin);
    		}
    		Collections.sort(bins);
    		return bins;
    	}
    	//排序算法
    	public void sort() {
    		List<Bin> bins = prepare();
    		while (true) {
    			// 取数组中最小的元素
    			MergeSource current = bins.get(0).source;
    			if (current.hasNext()) {// 从输入文件中取出下一个元素              
    				Bin newBin = new Bin(current, current.next());
    				// 二分查找，也就是和数组中已有元素进行比较
    				int index = Collections.binarySearch(bins, newBin);
    				if (index == 0 || index == -1) {// 算法思路情况1
    					this.out.write(newBin);
    					if (index == -1) {
    						throw new IllegalStateException("impossible");
    					}
    				} else {// 算法思路情况2
    					if (index < 0) {
    						index = -index - 1;
    					}
    					bins.add(index, newBin);
    					Bin minBin = bins.remove(0);
    					this.out.write(minBin);
    				}
    			} else {// 算法思路情况3:遇到文件尾
    				Bin minBin = bins.remove(0);
    				this.out.write(minBin);
    				if (bins.isEmpty()) {
    					break;
    				}
    			}
    		}
    	}
    	@Override
    	public void close() throws IOException {
    		for (MergeSource source : sources) {
    			source.close();
    		}
    		this.out.close();
    	}
    	//准备输入文件的内容
    	public static List<String> generateFiles(int n, int minEntries, int maxEntries) {
    		List<String> files = new ArrayList<>();
    		for (int i = 0; i < n; i++) {
    			String filename = "input-" + i + ".txt";
    			PrintWriter writer;
    			try {
    				writer = new PrintWriter(new FileOutputStream(filename));
    				int entries = ThreadLocalRandom.current().nextInt(minEntries, maxEntries);
    				List<Integer> nums = new ArrayList<>();
    				for (int k = 0; k < entries; k++) {
    					int num = ThreadLocalRandom.current().nextInt(10000000);
    					nums.add(num);
    				}
    				Collections.sort(nums);
    				for (int num : nums) {
    					writer.println(num);
    				}
    				writer.close();
    			} catch (FileNotFoundException e) {
    			}
    			files.add(filename);
    		}
    		return files;
    	}
    	public static void main(String[] args) throws IOException {
    		List<String> inputs = DiskMergeSort.generateFiles(100, 10000, 20000);
    		// 运行多次看算法耗时
    		for (int i = 0; i < 20; i++) {
    			DiskMergeSort sorter = new DiskMergeSort(inputs, "output.txt");
    			long start = System.currentTimeMillis();
    			sorter.sort();
    			long duration = System.currentTimeMillis() - start;
    			System.out.printf("%dms\n", duration);
    			sorter.close();
    		}
    	}
    }

**还有哪些缺陷？**

那就是如果输入文件数量非常多，那么内存中的数组就会特别大，对数组的插入删除操作肯定会很耗时，这时可以考虑使用 TreeSet 来代替数组。![90167984][]

[20190322222937942.png]: /images/20220316/5d30126f184f41ff85bf8cccb318c8e2.png
[20190322223658758.png]: /images/20220316/85b303e2c84346ac91e3f2792138d3d4.png
[20190322224438403.png]: /images/20220316/ee5577c22931473a81bc1154b8bd0e58.png
[20190322224545153.png]: /images/20220316/04aee479ef4246d4bfc6dd14d3d67701.png
[20190322224726411.png]: /images/20220316/a4b3389c995c4339920a45f970b54568.png
[20190322225436210.png]: /images/20220316/82f11ca07ad040b88b8d4ccd0fe4336b.png
[90167984]: /images/20220316/e34b37c5848944af802de3705c999051.png