Java 多线程并行处理大文件（ 淘宝2012校招技术笔试题 ）

 问题：统计一个单词可重复的英文文件（假设4G）中每个单词出现的次数，把结果按照英文排序放入一个文件中。并能够检索特定单词的出现次数。由于文件过大，不重复单词总数有限，需要考虑到执行速度和内存使用情况。（淘宝笔试技术题）

——————————————————————————————————————————————————————————

     import java.io.File;  

    import java.io.FileNotFoundException;  

    import java.io.FileOutputStream;  

    import java.io.IOException;  

    import java.io.RandomAccessFile;  

    import java.nio.ByteBuffer;  

    import java.nio.MappedByteBuffer;  

    import java.nio.channels.FileChannel;  

    import java.nio.channels.FileLock;  

    import java.nio.charset.Charset;  

    import java.util.HashMap;  

    import java.util.Map;  

    import java.util.StringTokenizer;  

    import java.util.TreeMap;  

      

    public class TestCountWords {  

        public static void main(String[] args) {  

            File wf = new File("words.txt");  

            final CountWords cw1 = new CountWords(wf, 0, wf.length()/2);  

            final CountWords cw2 = new CountWords(wf, wf.length()/2, wf.length());  

            final Thread t1 = new Thread(cw1);  

            final Thread t2 = new Thread(cw2);  

            //开辟两个线程分别处理文件的不同片段  

            t1.start();  

            t2.start();  

            Thread t = new Thread() {  

                public void run() {  

                    while(true) {  

                        //两个线程均运行结束  

                        if(Thread.State.TERMINATED==t1.getState() && Thread.State.TERMINATED==t2.getState()) {  

                            //获取各自处理的结果  

                            HashMap hMap1 = cw1.getResult();  

                            HashMap hMap2 = cw2.getResult();  

                            //使用TreeMap保证结果有序  

                            TreeMap tMap = new TreeMap();  

                            //对不同线程处理的结果进行整合  

                            tMap.putAll(hMap1);  

                            tMap.putAll(hMap2);  

                            //打印输出，查看结果  

                            for(Map.Entry entry : tMap.entrySet()) {  

                                String key = entry.getKey();    

                                int value = entry.getValue();    

                                System.out.println(key+":\t"+value);    

                            }  

                            //将结果保存到文件中  

                            mapToFile(tMap, new File("result.txt"));  

                        }  

                        return;  

                    }  

                }  

            };  

            t.start();  

        }  

        //将结果按照 "单词：次数" 格式存在文件中  

        private static void mapToFile(Map src, File dst) {  

            try {  

                //对将要写入的文件建立通道  

                FileChannel fcout = new FileOutputStream(dst).getChannel();  

                //使用entrySet对结果集进行遍历  

                for(Map.Entry entry : src.entrySet()) {  

                    String key = entry.getKey();  

                    int value = entry.getValue();  

                    //将结果按照指定格式放到缓冲区中  

                    ByteBuffer bBuf = ByteBuffer.wrap((key+":\t"+value).getBytes());  

                    fcout.write(bBuf);  

                    bBuf.clear();  

                }  

            } catch (FileNotFoundException e) {  

                e.printStackTrace();  

            } catch (IOException e) {  

                e.printStackTrace();  

            }  

        }  

    }  

      

    class CountWords implements Runnable {  

          

        private FileChannel fc;  

        private FileLock fl;  

        private MappedByteBuffer mbBuf;  

        private HashMap hm;  

          

        public CountWords(File src, long start, long end) {  

            try {  

                //得到当前文件的通道  

                fc = new RandomAccessFile(src, "rw").getChannel();  

                //锁定当前文件的部分  

                fl = fc.lock(start, end, false);  

                //对当前文件片段建立内存映射，如果文件过大需要切割成多个片段  

                mbBuf = fc.map(FileChannel.MapMode.READ_ONLY, start, end);  

                //创建HashMap实例存放处理结果  

                hm = new HashMap();  

            } catch (FileNotFoundException e) {  

                e.printStackTrace();  

            } catch (IOException e) {  

                e.printStackTrace();  

            }  

        }  

        @Override  

        public void run() {  

            String str = Charset.forName("UTF-8").decode(mbBuf).toString();  

            //使用StringTokenizer分析单词  

            StringTokenizer token = new StringTokenizer(str);  

            String word;  

            while(token.hasMoreTokens()) {  

                //将处理结果放到一个HashMap中，考虑到存储速度  

                word = token.nextToken();  

                if(null != hm.get(word)) {  

                    hm.put(word, hm.get(word)+1);  

                } else {  

                    hm.put(word, 1);  

                }  

            }  

            try {  

                //释放文件锁  

                fl.release();  

            } catch (IOException e) {  

                e.printStackTrace();  

            }  

            return;  

        }  

          

        //获取当前线程的执行结果  

        public HashMap getResult() {  

            return hm;  

        }  

    }  

以上代码的主要思想是：

1.使用具有键值对结构的HashMap来快速存取；

2.由于文件过大，用一个线程处理可能结果较慢，使用到并发机制；

3.IO操作比较耗时，所以使用了nio的相关内容；

4.最终结果要有序的话，可以使用TreeMap。

5.分开并行处理，最后合并结果，是不是有点 MapReduce 的味道呢？

类似问题有：

有10个1G大小的英文文件，每行中有一个英文单词，内存限制是1M，返回频数最高的100个单词。

循环一次将文件映射到内存中，每次映射1M，把处理结果放到一个文件中，最终将所有文件内容重新存入map中，对map合并，可得出结果。