公司的一道考试题算法分析——大数据量整数排序

    题目大意：移动公司需要对已经发放的所有139段的号码进行统计排序，已经发放的139号码段的文件都存放在一个文本文件中（原题是放在两个文件中），一个号码一行，现在需要将文件里的所有号码进行排序，并写入到一个新的文件中；号码可能会有很多，最多可能有一亿个不同的号码（所有的139段号码），存入文本文件中大概要占1.2G的空间；jvm最大的内存在300以内，程序要考虑程序的可执行性及效率；只能使用Java标准库，不得使用第三方工具。
    这是个典型的大数据量的排序算法问题，首先要考虑空间问题，一下把1.2G的数据读入内存是不太可能的，就算把1一亿条数据，转都转换成int类型存储也要占接近400M的空间。当时做个题目我并没有想太多的执行效率问题，主要就考虑了空间，而且习惯性的想到合并排序，基本思想是原文件分割成若干个小文件并排序，再将排序好的小文件合并得到最后结果，算法大概如下：

    1.顺序读取存放号码文件的中所有号码，并取139之后的八位转换为int类型；每读取号码数满一百万个（这个数据可配置）将已经读取的号码排序并存入新建的临时文件。
    2.将所有生成的号码有序的临时文件合并存入结果文件。

    这个算法虽然解决了空间问题，但是运行效率极低，由于IO读写操作太多，加上步骤1中的排序的算法（快速排序）本来效率就不高（对于电话排序这种特殊情况来说），导致1亿条数据排序运行3个小时才有结果。

    如果和能够减少排序的时间呢？首当其冲的减少IO操作，另外如果能够有更加好排序算法也行。前天无聊再看这个题目时突然想到大三时看《编程珠玑》时上面也有个问题的需求这个这个题目差不多，记得好像使用是位向量（实际上就是一个bit数组），用电话作为index，心中大喜，找到了解决此问题的最完美方案啦：用位向量存储电话号码，一个号码占一个bit，一亿个电话号码也只需要大概12M的空间；算法大概如下：
      1.初始化bits[capacity]；
      2.顺序所有读入电话号码，并转换为int类型，修改位向量值：bits[phoneNum]=1；
      3.遍历bits数组，如果bits[index]=1，转换index为电话号码输出。
    Java中没有bit类型，一个boolean值占空间为1byte（感兴趣的可以自己写程序验证），我自己写个个用int模拟bit数组的类，代码如下：
   

public class BitArray {
	private int[] bits = null;
	private int length;
	//用于设置或者提取int类型的数据的某一位(bit)的值时使用
	private final static int[] bitValue = {
		0x80000000,//10000000 00000000 00000000 00000000      
		0x40000000,//01000000 00000000 00000000 00000000      
		0x20000000,//00100000 00000000 00000000 00000000      
		0x10000000,//00010000 00000000 00000000 00000000      
		0x08000000,//00001000 00000000 00000000 00000000      
		0x04000000,//00000100 00000000 00000000 00000000      
		0x02000000,//00000010 00000000 00000000 00000000      
		0x01000000,//00000001 00000000 00000000 00000000      
		0x00800000,//00000000 10000000 00000000 00000000      
		0x00400000,//00000000 01000000 00000000 00000000      
		0x00200000,//00000000 00100000 00000000 00000000      
		0x00100000,//00000000 00010000 00000000 00000000      
		0x00080000,//00000000 00001000 00000000 00000000      
		0x00040000,//00000000 00000100 00000000 00000000      
		0x00020000,//00000000 00000010 00000000 00000000      
		0x00010000,//00000000 00000001 00000000 00000000		  
		0x00008000,//00000000 00000000 10000000 00000000      
		0x00004000,//00000000 00000000 01000000 00000000      
		0x00002000,//00000000 00000000 00100000 00000000      
		0x00001000,//00000000 00000000 00010000 00000000      
		0x00000800,//00000000 00000000 00001000 00000000      
		0x00000400,//00000000 00000000 00000100 00000000      
		0x00000200,//00000000 00000000 00000010 00000000      
		0x00000100,//00000000 00000000 00000001 00000000      
		0x00000080,//00000000 00000000 00000000 10000000      
		0x00000040,//00000000 00000000 00000000 01000000      
		0x00000020,//00000000 00000000 00000000 00100000      
		0x00000010,//00000000 00000000 00000000 00010000      
		0x00000008,//00000000 00000000 00000000 00001000      
		0x00000004,//00000000 00000000 00000000 00000100      
		0x00000002,//00000000 00000000 00000000 00000010      
		0x00000001 //00000000 00000000 00000000	00000001 			
	};
	public BitArray(int length) {
		if(length < 0){
			throw new IllegalArgumentException("length必须大于零！");
		}
		bits = new int[length / 32 + (length % 32 > 0 ? 1 : 0)];
		this.length = length;
	}
	//取index位的值
	public int getBit(int index){
		if(index <0 || index > length){
			throw new IllegalArgumentException("length必须大于零小于" + length);
		}
		int intData = bits[index/32];
		return (intData & bitValue[index%32]) >>> (32 - index%32 -1);
	}
	//设置index位的值，只能为0或者1
	public void setBit(int index,int value){
		if(index <0 || index > length){
			throw new IllegalArgumentException("length必须大于零小于" + length);
		}		
		if(value!=1&&value!=0){
			throw new IllegalArgumentException("value必须为0或者1");
		}
		int intData = bits[index/32];
		if(value == 1){
			bits[index/32] = intData | bitValue[index%32];
		}else{
			bits[index/32] = intData & ~bitValue[index%32];
		}
	}
	public int getLength(){
		return length;
	}	
}    
    

   
    bit数组有了，剩下就是算法代码，核心代码如下：
   

			bitArray = new BitArray(100000000);	
			//顺序读取所有的手机号码
			while((phoneNum = bufferedReader.readLine())!=null){
				phoneNum = phoneNum.trim().substring(3);//13573228432
				//取139后8位转换为int类型
				phoneNumAsInt = Integer.valueOf(phoneNum);
				//设置对应bit值为1
				bitArray.setBit(phoneNumAsInt, 1);
			}	
			//遍历bit数组输出所有存在的号码
			for(int i = 0;i<sortUnit;i++){
		    	if(bitArray.getBit(i)==1){
	    			writer.write("139" + leftPad(String.valueOf(i + sortUnit*times), 8));
	    			writer.newLine();			    		
		    	}				
			}
			writer.flush();
    

    经测试，修改后的算法排序时只需要20多M的内存，一亿条电话号码排序只要10分钟（时间主要花在IO上），看来效果还是很明显的。
    这个算法很快，不过也有他的局限性：
    1.只能用于整数的排序，或者可以准确映射到正整数（对象不同对应的正整数也不相同）的数据的排序。
    2.不能处理重复的数据，重复的数据排序后只有一条（如果有这种需求可以在这个算法的基础上修改，给出现次数大于1的数据添加个计数器，然后存入Map中）
    3.对于数据量极其大的数据处理可能还是比较占用空间，这种情况可配合多通道排序算法解决。

    PS:这个算法的思想源于《编程珠玑》，有兴趣的可以读读那本书，非常不错！
   

评论

12 楼 pxlfxl2 2012-08-12  

lavafree 写道

java的bit表示可用BitSet这样你的内存使用更小

思想一样的，占用内存也差不多，我这个绝不比BitSet多。

11 楼 lavafree 2012-07-24  

java的bit表示可用BitSet这样你的内存使用更小

10 楼 pxlfxl2 2011-07-08  

superobin 写道

大连软开的？

看来兄台也出自软开，哈哈……

9 楼 superobin 2011-07-07  

大连软开的？

8 楼 pxlfxl2 2010-10-08  

hanz188 写道

终于知道这个算法叫什么名字了：布隆过滤器（Bloom Filter），详细说明可以在这里找到：http://www.cnblogs.com/kevinyang/archive/2009/02/01/1381803.html

嗯，思想是一样的，用二进向量来存储数据，达到减少占用存储空间；不过在排序中就得充分利用向量的index值了，而且使用场景是非常少。

7 楼 hanz188 2010-10-08  

终于知道这个算法叫什么名字了：布隆过滤器（Bloom Filter），详细说明可以在这里找到：http://www.cnblogs.com/kevinyang/archive/2009/02/01/1381803.html

6 楼 pxlfxl2 2010-09-21  

hanz188 写道

pxlfxl2 写道

e_soft 写道

类似一个排序好的大map，穷举了所有电话号码，然后有电话号码就把标志位设为1，
最后输出所有标志位为1的电话

嗯，可以这么理解，但是采用了bit数组而不是map，bit数组的index对应着电话号码，对应的值（只能为0或者1）表示这个电话号码否存在，一个电话号码只占一个bit，这样的大大节省了存储空间；由于相当于直接拿电话号码作为bit数组的index（在这里是去139后八位），所以只需顺序遍历bit数组，转换bit值为1的index到对应的手机号输出就是有序的啦；算法的时间复杂度是o(n)。

我明白了，现在还想到一个问题，如果我们现在要处理的不是11位的手机号码，而是其它号码，比如说身份证呢，假设身份证号码全是18位，那怎么将这个将身份证号码转换成位向量，byte类型只有8位，肯定不够用，用int类型？

哈哈，你还是没看明白，跟byte和int类型关系不大，这里用的的bit数组的index去表示电话号码。
在最后面我提到过这个算法的局限性，只是在特定的情况下才会使用的。

5 楼 hanz188 2010-09-21  

pxlfxl2 写道

e_soft 写道

类似一个排序好的大map，穷举了所有电话号码，然后有电话号码就把标志位设为1，
最后输出所有标志位为1的电话

嗯，可以这么理解，但是采用了bit数组而不是map，bit数组的index对应着电话号码，对应的值（只能为0或者1）表示这个电话号码否存在，一个电话号码只占一个bit，这样的大大节省了存储空间；由于相当于直接拿电话号码作为bit数组的index（在这里是去139后八位），所以只需顺序遍历bit数组，转换bit值为1的index到对应的手机号输出就是有序的啦；算法的时间复杂度是o(n)。

我明白了，现在还想到一个问题，如果我们现在要处理的不是11位的手机号码，而是其它号码，比如说身份证呢，假设身份证号码全是18位，那怎么将这个将身份证号码转换成位向量，byte类型只有8位，肯定不够用，用int类型？

4 楼 pxlfxl2 2010-09-21  

e_soft 写道

类似一个排序好的大map，穷举了所有电话号码，然后有电话号码就把标志位设为1，
最后输出所有标志位为1的电话

嗯，可以这么理解，但是采用了bit数组而不是map，bit数组的index对应着电话号码，对应的值（只能为0或者1）表示这个电话号码否存在，一个电话号码只占一个bit，这样的大大节省了存储空间；由于相当于直接拿电话号码作为bit数组的index（在这里是去139后八位），所以只需顺序遍历bit数组，转换bit值为1的index到对应的手机号输出就是有序的啦；算法的时间复杂度是o(n)。

3 楼 e_soft 2010-09-21  

类似一个排序好的大map，穷举了所有电话号码，然后有电话号码就把标志位设为1，
最后输出所有标志位为1的电话

2 楼 pxlfxl2 2010-09-21  

hanz188 写道

这个文章我没有看明白，准确的说，是算法的描述没有看明白。怎么把电话号码转换成位向量呢？

这个后面的那以小段程序写的很明白啦，因为只有139号码段的号码，取手机号码的后八位作为bit数组的index就可以啦

1 楼 hanz188 2010-09-21  

这个文章我没有看明白，准确的说，是算法的描述没有看明白。怎么把电话号码转换成位向量呢？