使用Fork/Join实现WordCount例子

素颜马尾好姑娘i 2022-09-26 03:50 153阅读 0赞

Fork/Join框架是Java7提供了的一个用于并行执行任务的框架， 是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。  
Fork/Join框架是一个比较特殊的线程池框架，专用于需要将一个任务不断分解成子任务（分叉），再不断进行汇总得到最终结果（结合）的计算过程。  
比起传统的线程池类ThreadPoolExecutor，ForkJoinPool 实现了工作窃取算法，使得空闲线程能够主动分担从别的线程分解出来的子任务，  
从而让所有的线程都尽可能处于饱满的工作状态，提高执行效率。

下面将演示一个使用Fork/Join实现WordCount例子，假设有一个目录，里面都是文本文件，现在要统计这些所有的文件里面的字符串的个数

package com.fork;
    
    import java.io.File;
    import java.io.IOException;
    import java.nio.file.Files;
    import java.nio.file.Paths;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.RecursiveTask;
    
    /**
     * RecursiveTask 表示有返回值的计算
     */
    public class FileRecursiveTask extends RecursiveTask<Map<String, Integer>> {
    
    	private static final long serialVersionUID = 1L;
    	private final List<String> contents;
    	
    	public FileRecursiveTask(File file) {
    		System.out.println("处理文件：" + file.getAbsolutePath());
    		try {
    			contents = Files.readAllLines(Paths.get(file.toURI()));
    		} catch (IOException e) {
    			throw new RuntimeException(e);
    		}
    	}
    	
    	/**
    	 * 业务逻辑处理，相当于map
    	 */
    	protected Map<String, Integer> compute() {
    		Map<String, Integer> map = new HashMap<>();
    		for(String content : contents){
    			String[] strs = content.split(" ");
    			for(String str : strs){
    				if(map.containsKey(str)){
    					int val = map.get(str);
    					map.put(str, val+1);
    				} else {
    					map.put(str, 1);
    				}
    			}
    		}
    		return map;
    	}
    }

package com.fork;
    
    import java.io.File;
    import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.ForkJoinTask;
    import java.util.concurrent.RecursiveTask;
    
    /**
     * 每个文件会使用一个线程去处理，最终汇总到一起
     */
    public class ForkRecursiveTask extends RecursiveTask<Map<String, Integer>> {
    
    	private static final long serialVersionUID = 1L;
    	private final File[] files;
    	
    	public ForkRecursiveTask(String path) {
    		files = new File(path).listFiles();
    	}
    	
    	/**
    	 * 汇总处理，相当于reduce
    	 */
    	protected Map<String, Integer> compute() {
    		List<ForkJoinTask<Map<String, Integer>>> tasks = new ArrayList<>();
    		for(File file : files){
    			FileRecursiveTask frt = new FileRecursiveTask(file);
    			tasks.add(frt.fork());
    		}
    		
    		Map<String, Integer> result = new HashMap<>();
    		for(ForkJoinTask<Map<String, Integer>> task : tasks){
    			Map<String, Integer> map = task.join();
    			for(String key : map.keySet()){
    				if(result.containsKey(key)){
    					result.put(key, result.get(key) + map.get(key));
    				} else {
    					result.put(key, map.get(key));
    				}
    			}
    		}
    		return result;
    	}
    }

package com.fork;
    
    import java.util.Map;
    import java.util.concurrent.ForkJoinPool;
    
    public class ForkJoinPoolTest {
    	public static void main(String[] args) {
    		ForkJoinPool forkJoinPool = new ForkJoinPool();
    		Map<String, Integer> map = forkJoinPool.invoke(new ForkRecursiveTask("e:/txts"));
    		//输出最终所有的计算结果
    		for(String key : map.keySet()){
    			System.out.println(key + "=" + map.get(key));
    		}
    	}
    }