ITPub博客

首页 > 应用开发 > Java > 外部排序

外部排序

原创 Java 作者:壹頁書 时间:2014-05-23 00:15:29 0 删除 编辑
外部排序
如果需要排序的文件超过了内存的大小,就需要分块排序,按块写入文件,最后再归并已经有序的分块文件。

第一阶段 切分排序
1.将原始未排序的大文件,分为有序的小文件
    1.1确定大文件切分的位置
    1.2主线程按块读取原始文件,然后交给排序线程
    1.3排序线程,对块进行排序
    1.4写入线程将有序的数据,写入文件


第二阶段 归并
    将临时文件归并到目标的结果文件

作为JAVA程序,需要注意的是,第一阶段是IO密集型,尽量分配更多的内存给Old区
而第二阶段是CPU密集型,需要注意Young区,避免过多的GC占用CPU资源。

需要提前了解的技术细节。
1.JAVA切分大文件
http://blog.itpub.net/29254281/viewspace-1161173/

2.JAVA内存映射文件
http://blog.itpub.net/29254281/viewspace-1162157/

3.JAVA栅栏
http://blog.itpub.net/29254281/viewspace-1164727/

4.观察者模式、生产者/消费者模式    

5.JVM监控和GC

实验环境:
双核CPU,使用1G的JAVA堆内存,对2亿long型随机数排序,一般情况下,2亿long型随机数文件4G左右大小。




总体设计的思路
Main主线程负责切分原始文件,
blocking函数根据文件分片的设置,返回原始文件的分块。
这样主要为了分块包含整行,而不会切断数据。
读取出来的原始未排序数据封装为Sorter对象,提交给排序线程池。
排序之后的数据,封装为Writer对象,再交给写入线程池,将排序好的分块写入临时文件。
在每个分块写入之后,线程在CyclicBarrier栅栏处等待,待所有的分块写入完毕,启用归并。
需要特别注意的是,在栅栏等待之前,一定要释放所有的资源,以便JVM GC回收内存。
就是Writer对象write方法的下列代码

归并非常消耗CPU资源。
在归并的过程中,因为每个分块本身都是有序的,所以只需要一个线程计算各个分块中最小的数字,将其写入BlockingQueue。
而另一个线程,不断的将队列的数据顺序写入目标文件。
这就是Merge对象的作用。这个过程使用了观察者模式和生产者消费者模式。

实现如下:
  1. import java.io.BufferedInputStream;
  2. import java.io.BufferedOutputStream;
  3. import java.io.File;
  4. import java.io.FileInputStream;
  5. import java.io.FileOutputStream;
  6. import java.io.IOException;
  7. import java.io.RandomAccessFile;
  8. import java.nio.MappedByteBuffer;
  9. import java.nio.channels.FileChannel;
  10. import java.nio.channels.FileChannel.MapMode;
  11. import java.util.ArrayList;
  12. import java.util.Arrays;
  13. import java.util.Collections;
  14. import java.util.List;
  15. import java.util.Queue;
  16. import java.util.concurrent.BlockingQueue;
  17. import java.util.concurrent.BrokenBarrierException;
  18. import java.util.concurrent.ConcurrentLinkedQueue;
  19. import java.util.concurrent.CyclicBarrier;
  20. import java.util.concurrent.ExecutorService;
  21. import java.util.concurrent.Executors;
  22. import java.util.concurrent.LinkedBlockingQueue;

  23. public class Controller {
  24.     public static void main(String[] args) throws IOException {
  25.         Controller c = new Controller(new File("/home/lihuilin/桌面/t.txt"), 15, "/home/lihuilin/桌面/");
  26.     }

  27.     // 排序线程池
  28.     private final ExecutorService sortThread;
  29.     // 写入线程池,将排序之后的分片写入文件
  30.     private final ExecutorService writerThread;
  31.     // 栅栏,等待所有分片写入文件之后,启动合并
  32.     private final CyclicBarrier barrier;
  33.     // 原始没有顺序的大文件
  34.     private final File file;
  35.     // 分片大小
  36.     private final int pieces;
  37.     // 输出目录
  38.     private final String outDir;
  39.     // 记录分片写入临时文件的位置
  40.     private final List<File> outFileList = new ArrayList<File>();

  41.     public Controller(File file, int pieces, final String outDir) throws IOException {
  42.         final long start = System.currentTimeMillis();
  43.         sortThread = Executors.newFixedThreadPool(1);
  44.         // 写入线程池的线程数一定不能小于分片的大小。否则CyclicBarrier
  45.         // await之后,后续的分片将没有线程可用。await不会释放线程资源。
  46.         writerThread = Executors.newFixedThreadPool(pieces + 1);
  47.         this.file = file;
  48.         this.pieces = pieces;
  49.         this.outDir = outDir;
  50.         this.barrier = new CyclicBarrier(pieces, new Runnable() {

  51.             @Override
  52.             public void run() {
  53.                 long end = System.currentTimeMillis();
  54.                 System.out.println("合并之前总用时:" + (end - start) / 1000 + "s");
  55.                 // 合并有序的分片临时文件
  56.                 Merger merger = new Merger(outFileList, outDir);
  57.                 writerThread.submit(merger);
  58.                 try {
  59.                     merger.merge();
  60.                 } catch (IOException e) {
  61.                     // TODO Auto-generated catch block
  62.                     e.printStackTrace();
  63.                 } catch (InterruptedException e) {
  64.                     // TODO Auto-generated catch block
  65.                     e.printStackTrace();
  66.                 }

  67.                 writerThread.shutdown();
  68.                 sortThread.shutdown();
  69.                 end = System.currentTimeMillis();
  70.                 System.out.println("外部排序总用时:" + (end - start) / 1000 + "s");
  71.             }
  72.         });
  73.         action();
  74.     }

  75.     private void action() throws IOException {
  76.         List<Point> list = blocking(file, pieces);
  77.         for (Point p : list) {
  78.             Spilter spilter = new MappedByteBufferSpilter(file, p);
  79.             long[] data = null;
  80.             data = spilter.spilt();
  81.             Sorter s = new Sorter(data, p, writerThread, barrier, outFileList);
  82.             sortThread.submit(s);
  83.         }

  84.     }

  85.     private List<Point> blocking(File file, int piece) throws IOException {
  86.         List<Point> result = new ArrayList<Point>();
  87.         List<Long> list = new ArrayList<Long>();
  88.         list.add(-1L);
  89.         long length = file.length();
  90.         long step = length / piece;
  91.         long index = 0;
  92.         for (int i = 0; i < piece; i++) {
  93.             BufferedInputStream in = new BufferedInputStream(new FileInputStream(file));
  94.             if (index + step < length) {
  95.                 index = index + step;
  96.                 in.skip(index);

  97.                 while (in.read() != 10) {
  98.                     index = index + 1;
  99.                 }
  100.                 list.add(index);
  101.                 index++;
  102.             }
  103.             in.close();
  104.         }
  105.         list.add(length - 1);
  106.         for (int i = 0; i < list.size() - 1; i++) {
  107.             long skipSize = list.get(i) + 1;
  108.             long l = list.get(i + 1) - list.get(i);
  109.             result.add(new Point(skipSize, l, outDir));

  110.         }
  111.         return result;
  112.     }

  113. }

  114. class Merger implements Runnable {
  115.     private final List<Worker> workerList = new ArrayList<Worker>();
  116.     private String outDir = null;
  117.     private BlockingQueue<Long> queue = new LinkedBlockingQueue<Long>(1000);
  118.     private volatile boolean finished = false;

  119.     public Merger(List<File> outFileList, String outDir) {
  120.         for (File file : outFileList) {
  121.             Worker worker = new Worker(file, workerList);
  122.             workerList.add(worker);
  123.         }
  124.         this.outDir = outDir;
  125.     }

  126.     @Override
  127.     public void run() {
  128.         try {
  129.             System.out.println("读取队列,写入目标文件");
  130.             BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(outDir + "result.txt"), 50 * 1024 * 1024);
  131.             while (finished != true || !queue.isEmpty()) {
  132.                 Long l = queue.take();
  133.                 bos.write((l + "\n").getBytes());
  134.             }
  135.             bos.flush();
  136.             bos.close();
  137.         } catch (Exception ex) {
  138.             ex.printStackTrace();
  139.         }
  140.     }

  141.     public void merge() throws IOException, InterruptedException {
  142.         while (workerList.size() != 0) {
  143.             Collections.sort(workerList);
  144.             Worker worker = workerList.get(0);
  145.             Long data = worker.poll();
  146.             if (data == null) {
  147.                 workerList.remove(worker);
  148.             } else {
  149.                 queue.put(data);
  150.             }
  151.         }
  152.         finished = true;
  153.     }

  154.     private class Worker implements Comparable<Worker> {
  155.         private long data;
  156.         private MappedByteBuffer buffer = null;
  157.         private List<Worker> workerList = null;
  158.         private boolean eof = false;

  159.         Worker(File file, List<Worker> workerList) {
  160.             try {
  161.                 RandomAccessFile rFile = new RandomAccessFile(file, "r");
  162.                 FileChannel channel = rFile.getChannel();
  163.                 buffer = channel.map(MapMode.READ_ONLY, 0, channel.size());
  164.                 channel.close();
  165.                 rFile.close();
  166.                 this.workerList = workerList;
  167.                 data = buffer.getLong();
  168.             } catch (IOException e) {
  169.                 // TODO Auto-generated catch block
  170.                 e.printStackTrace();
  171.             }
  172.         }

  173.         public long peek() {
  174.             return data;
  175.         }

  176.         public Long poll() {
  177.             long result = data;
  178.             if (buffer.position() != buffer.limit()) {
  179.                 data = buffer.getLong();
  180.             } else {
  181.                 if (eof == false) {
  182.                     eof = true;
  183.                 } else {
  184.                     return null;
  185.                 }
  186.             }

  187.             return result;
  188.         }

  189.         @Override
  190.         public int compareTo(Worker o) {
  191.             if (this.peek() > o.peek()) {
  192.                 return 1;
  193.             } else if (this.peek() < o.peek()) {
  194.                 return -1;
  195.             } else {
  196.                 return 0;
  197.             }
  198.         }
  199.     }

  200. }

  201. interface Spilter {
  202.     public long[] spilt();
  203. }

  204. class Sorter implements Runnable {
  205.     long[] data;
  206.     Point p;
  207.     ExecutorService writerThread;
  208.     List<File> outFileList;
  209.     CyclicBarrier barrier;

  210.     public Sorter(long[] data, Point p, ExecutorService writerThread, CyclicBarrier barrier, List<File> outFileList) {
  211.         this.data = data;
  212.         this.p = p;
  213.         this.outFileList = outFileList;
  214.         this.barrier = barrier;
  215.         this.writerThread = writerThread;
  216.     }

  217.     public long[] sort() {
  218.         System.out.println("\t开始排序:" + p);
  219.         long start = System.currentTimeMillis();
  220.         Arrays.sort(this.data);
  221.         long end = System.currentTimeMillis();
  222.         System.out.println("\t结束排序:" + p + ",用时:" + (end - start) / 1000);
  223.         return this.data;
  224.     }

  225.     @Override
  226.     public void run() {
  227.         Writer writer = new MappedByteBufferWriter(sort(), p, barrier, outFileList);
  228.         writerThread.submit(writer);
  229.     }
  230. }

  231. interface Writer extends Runnable {
  232.     public void write();
  233. }

  234. class MappedByteBufferWriter implements Writer {
  235.     private static int FLAG = 1;
  236.     private CyclicBarrier barrier = null;
  237.     private File outfile = null;
  238.     private Point point = null;

  239.     private long[] data = null;
  240.     private List<File> outFileList = null;

  241.     public MappedByteBufferWriter(long[] data, Point point, CyclicBarrier barrier, List<File> outFileList) {
  242.         this.data = data;
  243.         this.point = point;
  244.         this.outfile = new File(point.getOutDir() + FLAG + ".txt");
  245.         this.barrier = barrier;
  246.         this.outFileList = outFileList;
  247.         FLAG++;
  248.     }

  249.     @Override
  250.     public void write() {
  251.         try {
  252.             System.out.println("\t\t开始写入:" + point);
  253.             long start = System.currentTimeMillis();
  254.             FileChannel channel = new RandomAccessFile(this.outfile, "rw").getChannel();
  255.             MappedByteBuffer buffer = channel.map(MapMode.READ_WRITE, 0, this.data.length * 8);
  256.             for (int i = 0; i < data.length; i++) {
  257.                 buffer.putLong(data[i]);
  258.             }
  259.             buffer.force();
  260.             long end = System.currentTimeMillis();
  261.             System.out.println("\t\t结束写入:" + point + ",用时:" + (end - start) / 1000);
  262.             synchronized (outFileList) {
  263.                 outFileList.add(outfile);
  264.             }
  265.             this.data = null;
  266.             channel.close();
  267.             buffer = null;
  268.             barrier.await();
  269.         } catch (IOException ex) {
  270.             ex.printStackTrace();
  271.         } catch (InterruptedException e) {
  272.             // TODO Auto-generated catch block
  273.             e.printStackTrace();
  274.         } catch (BrokenBarrierException e) {
  275.             // TODO Auto-generated catch block
  276.             e.printStackTrace();
  277.         }
  278.     }

  279.     @Override
  280.     public void run() {
  281.         this.write();
  282.     }

  283. }

  284. class MappedByteBufferSpilter implements Spilter {
  285.     private File file;
  286.     private Point point;

  287.     public MappedByteBufferSpilter(File file, Point p) {
  288.         this.file = file;
  289.         this.point = p;
  290.     }

  291.     @Override
  292.     public long[] spilt() {
  293.         System.out.println("开始读入:" + point);
  294.         long start = System.currentTimeMillis();
  295.         long[] result = null;
  296.         try {

  297.             FileChannel in = new RandomAccessFile(file, "r").getChannel();
  298.             MappedByteBuffer inBuffer = in.map(MapMode.READ_ONLY, point.getSkipSize(), point.getLength());

  299.             byte[] data = new byte[inBuffer.limit()];
  300.             inBuffer.get(data);
  301.             result = new long[getObjectSize(data)];
  302.             int resultIndex = 0;
  303.             int index = 0;
  304.             int first = 0;
  305.             while (index < data.length) {
  306.                 if (data[index] == 10) {
  307.                     byte[] tmpData = Arrays.copyOfRange(data, first, index);
  308.                     String str = new String(tmpData);
  309.                     result[resultIndex] = Long.valueOf(str);
  310.                     resultIndex++;
  311.                     first = index + 1;
  312.                 }

  313.                 index++;
  314.             }
  315.             in.close();

  316.         } catch (IOException ex) {
  317.             ex.printStackTrace();
  318.         }
  319.         long end = System.currentTimeMillis();
  320.         System.out.println("结束读入:" + point + ",用时:" + (end - start) / 1000);
  321.         return result;
  322.     }

  323.     private int getObjectSize(byte[] data) {
  324.         int size = 0;
  325.         for (byte b : data) {
  326.             if (b == 10) {
  327.                 size++;
  328.             }
  329.         }
  330.         return size;
  331.     }

  332. }

  333. class Point {
  334.     public Point(long skipSize, long length, String outDir) {
  335.         if (length > Integer.MAX_VALUE) {
  336.             throw new RuntimeException("长度溢出");
  337.         }
  338.         this.skipSize = skipSize;
  339.         this.length = (int) length;
  340.         this.outDir = outDir;
  341.     }

  342.     @Override
  343.     public String toString() {
  344.         return "Point [skipSize=" + skipSize + ", length=" + length + "]";
  345.     }

  346.     private long skipSize;
  347.     private int length;
  348.     private String outDir;

  349.     public String getOutDir() {
  350.         return outDir;
  351.     }

  352.     public long getSkipSize() {
  353.         return skipSize;
  354.     }

  355.     public int getLength() {
  356.         return length;
  357.     }
  358. }
运行:
  1. [lihuilin@lihuilin 桌面]$ java Controller
  2. 开始读入:Point [skipSize=0, length=271726519]
  3. 结束读入:Point [skipSize=0, length=271726519],用时:8
  4. 开始读入:Point [skipSize=271726519, length=271726515]
  5. 开始排序:Point [skipSize=0, length=271726519]
  6. 结束排序:Point [skipSize=0, length=271726519],用时:3
  7. 开始写入:Point [skipSize=0, length=271726519]
  8. 结束写入:Point [skipSize=0, length=271726519],用时:2
  9. 结束读入:Point [skipSize=271726519, length=271726515],用时:9
  10. 开始排序:Point [skipSize=271726519, length=271726515]
  11. 开始读入:Point [skipSize=543453034, length=271726511]
  12. 结束排序:Point [skipSize=271726519, length=271726515],用时:4
  13. 开始写入:Point [skipSize=271726519, length=271726515]
  14. 结束写入:Point [skipSize=271726519, length=271726515],用时:3
  15. 结束读入:Point [skipSize=543453034, length=271726511],用时:9
  16. 开始读入:Point [skipSize=815179545, length=271726515]
  17. 开始排序:Point [skipSize=543453034, length=271726511]
  18. 结束排序:Point [skipSize=543453034, length=271726511],用时:3
  19. 开始写入:Point [skipSize=543453034, length=271726511]
  20. 结束写入:Point [skipSize=543453034, length=271726511],用时:5
  21. 结束读入:Point [skipSize=815179545, length=271726515],用时:13
  22. 开始读入:Point [skipSize=1086906060, length=271726524]
  23. 开始排序:Point [skipSize=815179545, length=271726515]
  24. 结束排序:Point [skipSize=815179545, length=271726515],用时:3
  25. 开始写入:Point [skipSize=815179545, length=271726515]
  26. 结束写入:Point [skipSize=815179545, length=271726515],用时:5
  27. 结束读入:Point [skipSize=1086906060, length=271726524],用时:13
  28. 开始读入:Point [skipSize=1358632584, length=271726507]
  29. 开始排序:Point [skipSize=1086906060, length=271726524]
  30. 结束排序:Point [skipSize=1086906060, length=271726524],用时:3
  31. 开始写入:Point [skipSize=1086906060, length=271726524]
  32. 结束写入:Point [skipSize=1086906060, length=271726524],用时:5
  33. 结束读入:Point [skipSize=1358632584, length=271726507],用时:12
  34. 开始读入:Point [skipSize=1630359091, length=271726523]
  35. 开始排序:Point [skipSize=1358632584, length=271726507]
  36. 结束排序:Point [skipSize=1358632584, length=271726507],用时:3
  37. 开始写入:Point [skipSize=1358632584, length=271726507]
  38. 结束写入:Point [skipSize=1358632584, length=271726507],用时:5
  39. 结束读入:Point [skipSize=1630359091, length=271726523],用时:13
  40. 开始读入:Point [skipSize=1902085614, length=271726514]
  41. 开始排序:Point [skipSize=1630359091, length=271726523]
  42. 结束排序:Point [skipSize=1630359091, length=271726523],用时:3
  43. 开始写入:Point [skipSize=1630359091, length=271726523]
  44. 结束写入:Point [skipSize=1630359091, length=271726523],用时:5
  45. 结束读入:Point [skipSize=1902085614, length=271726514],用时:13
  46. 开始读入:Point [skipSize=2173812128, length=271726519]
  47. 开始排序:Point [skipSize=1902085614, length=271726514]
  48. 结束排序:Point [skipSize=1902085614, length=271726514],用时:3
  49. 开始写入:Point [skipSize=1902085614, length=271726514]
  50. 结束写入:Point [skipSize=1902085614, length=271726514],用时:5
  51. 结束读入:Point [skipSize=2173812128, length=271726519],用时:13
  52. 开始读入:Point [skipSize=2445538647, length=271726516]
  53. 开始排序:Point [skipSize=2173812128, length=271726519]
  54. 结束排序:Point [skipSize=2173812128, length=271726519],用时:3
  55. 开始写入:Point [skipSize=2173812128, length=271726519]
  56. 结束写入:Point [skipSize=2173812128, length=271726519],用时:5
  57. 结束读入:Point [skipSize=2445538647, length=271726516],用时:12
  58. 开始读入:Point [skipSize=2717265163, length=271726517]
  59. 开始排序:Point [skipSize=2445538647, length=271726516]
  60. 结束排序:Point [skipSize=2445538647, length=271726516],用时:3
  61. 开始写入:Point [skipSize=2445538647, length=271726516]
  62. 结束写入:Point [skipSize=2445538647, length=271726516],用时:5
  63. 结束读入:Point [skipSize=2717265163, length=271726517],用时:13
  64. 开始读入:Point [skipSize=2988991680, length=271726517]
  65. 开始排序:Point [skipSize=2717265163, length=271726517]
  66. 结束排序:Point [skipSize=2717265163, length=271726517],用时:3
  67. 开始写入:Point [skipSize=2717265163, length=271726517]
  68. 结束写入:Point [skipSize=2717265163, length=271726517],用时:5
  69. 结束读入:Point [skipSize=2988991680, length=271726517],用时:12
  70. 开始读入:Point [skipSize=3260718197, length=271726516]
  71. 开始排序:Point [skipSize=2988991680, length=271726517]
  72. 结束排序:Point [skipSize=2988991680, length=271726517],用时:3
  73. 开始写入:Point [skipSize=2988991680, length=271726517]
  74. 结束写入:Point [skipSize=2988991680, length=271726517],用时:5
  75. 结束读入:Point [skipSize=3260718197, length=271726516],用时:12
  76. 开始读入:Point [skipSize=3532444713, length=271726515]
  77. 开始排序:Point [skipSize=3260718197, length=271726516]
  78. 结束排序:Point [skipSize=3260718197, length=271726516],用时:3
  79. 开始写入:Point [skipSize=3260718197, length=271726516]
  80. 结束写入:Point [skipSize=3260718197, length=271726516],用时:5
  81. 结束读入:Point [skipSize=3532444713, length=271726515],用时:12
  82. 开始读入:Point [skipSize=3804171228, length=271726376]
  83. 开始排序:Point [skipSize=3532444713, length=271726515]
  84. 结束排序:Point [skipSize=3532444713, length=271726515],用时:3
  85. 开始写入:Point [skipSize=3532444713, length=271726515]
  86. 结束写入:Point [skipSize=3532444713, length=271726515],用时:4
  87. 结束读入:Point [skipSize=3804171228, length=271726376],用时:12
  88. 开始排序:Point [skipSize=3804171228, length=271726376]
  89. 结束排序:Point [skipSize=3804171228, length=271726376],用时:2
  90. 开始写入:Point [skipSize=3804171228, length=271726376]
  91. 结束写入:Point [skipSize=3804171228, length=271726376],用时:3
  92. 合并之前总用时:190s
  93. 读取队列,写入目标文件
  94. 外部排序总用时:398s
JVM监控:


从监控可以看到第一阶段是IO密集型,对于内存需要很大;
而第二阶段,排序各个分块对于CPU压力很大,一定要注意不要让GC线程占用过多CPU资源,就是Young区不能过小。

上图中FGC 55之后的就是第二阶段的过程,Young GC明显增多。

验证:
开始使用1-100的实验数据,可以正确排序。
大文件排序之后,可以使用Linux Sort命令验证。


关于性能优化
外部排序的本质就是排序写入小文件,再将小文件归并为有序的目标文件。
所以时间大致应该是拷贝这个文件的时间乘以2.

但是...
想起在15所的时候,吴老师使用大致与我性能相当的配置,居然只用了230S左右..
优化需要注意的两点
1.内存映射文件
    文件复制,分块文件写入然后读出这种场景使用内存映射文件,避免了内核空间和用户空间的内存复制。并且可以使用堆外操作系统内存作为缓存.
2.避免GC占用CPU
    读取文件和写入文件的时候,Byte和Long做转换的时候,均采用了String类型作为中转,
    

    
    后续可以考虑直接将byte和long型互转,避免引入String类型,这样就可以避免额外的GC

第一点都很容易想到,而第二点吴老师做了,我没有实现,这可能是导致性能较慢的原因。
以后有时间再补上这个细节吧。
因为使用了内存映射文件,比了避免误差,每次实验之前最好清除缓存。


来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/29254281/viewspace-1167988/ ,如需转载,请注明出处,否则将追究法律责任。

上一篇: MySQL的Group By分组
下一篇: 外部排序优化
请登录后发表评论 登录
全部评论

注册时间:2013-10-19

  • 博文量
    620
  • 访问量
    6339971