java多线程之分治技术框架Fork/Join

超、凢脫俗 2022-05-29 11:43 80阅读 0赞

一、Fork/Join框架用来解决能够通过分治技术将问题拆分成小任务的问题。例如某商城需要统计1百万个用户的消费总金额，如果采用顺序执行，一个一个去计算用户的消费金额，然后再合并计算，那么将非常耗时。此时可以用Fork/Join分治技术很好的实现多任务并行计算用户的消费金额，节省大量的时间。（但将消耗更多的资源）

二、Fork/Join框架之RecursiveAction（没有返回结果的任务）

package com.np.ota.test.forkjoin;
    
    public class Product {
    
    	private String name;
    	
    	private double price;
    
    	public String getName() {
    		return name;
    	}
    
    	public void setName(String name) {
    		this.name = name;
    	}
    
    	public double getPrice() {
    		return price;
    	}
    
    	public void setPrice(double price) {
    		this.price = price;
    	}
    	
    }

package com.np.ota.test.forkjoin;
    
    import java.util.ArrayList;
    import java.util.List;
    
    public class ProductListGenerator {
    
    	public List<Product> generate(int size){
    		List<Product> ret = new ArrayList<Product>();
    		for(int i = 0; i < size; i++){
    			Product product = new Product();
    			product.setName("product "+i);
    			product.setPrice(10);
    			ret.add(product);
    		}
    		return ret;
    	}
    	
    }

package com.np.ota.test.forkjoin;
    
    import java.util.List;
    import java.util.concurrent.RecursiveAction;
    
    public class Task extends RecursiveAction{
    
    	private static final long serialVersionUID = 1L;
    	
    	private List<Product> products;
    
    	private int first;
    	private int last;
    	private double increment;
    	
    	public Task(List<Product> products, int first, int last, double increment) {
    		this.products = products;
    		this.first = first;
    		this.last = last;
    		this.increment = increment;
    	}
    
    	@Override
    	protected void compute() {
    		if(last - first < 10){
    			updatePrices();
    		}else{
    			int middle = (last+first)/2;
    			System.out.println("pending tasks : "+getQueuedTaskCount());
    			Task t1 = new Task(products, first, middle+1, increment);
    			Task t2 = new Task(products, middle+1, last, increment);
    			invokeAll(t1, t2);
    		}
    	}
    
    	/**
    	 * 每个产品涨价increment
    	 */
    	private void updatePrices() {
    		for(int i = first; i < last; i++){
    			Product product = products.get(i);
    			product.setPrice(product.getPrice()*(1+increment));
    		}
    	}
    
    }

package com.np.ota.test.forkjoin;
    
    import java.util.List;
    import java.util.concurrent.ForkJoinPool;
    
    public class Main {
    
    	public static void main(String[] args) {
    
    		//创建一万个产品
    		ProductListGenerator generator = new ProductListGenerator();
    		List<Product> products = generator.generate(10000);
    		
    		//执行增加价格任务
    		Task task = new Task(products, 0, products.size(), 0.20);
    		ForkJoinPool pool = new ForkJoinPool();
    		pool.execute(task);
    		
    		//等待任务执行完毕
    		while(!task.isDone()){
    			System.out.println("ActiveThread:"+pool.getActiveThreadCount());
    		}
    		pool.shutdown();
    		
    		System.out.println("------------------------------");
    		for(int i = 0; i < products.size(); i++){
    			Product product = products.get(i);
    			if(product.getPrice() != 12){
    				System.out.println(product.getName());
    			}
    		}
    		
    	}
    
    }

三、Fork/Join框架之RecursiveTask（有返回结果的任务）

package com.np.ota.test.forkjoin2;
    
    import java.util.concurrent.RecursiveTask;
    
    public class Sum extends RecursiveTask<Long>{
    
    	private static final long serialVersionUID = 1L;
    	
    	private long start;
    	private long end;
    	private long threshold;
    	private long[] numbers;
    	
    	public Sum(long start, long end, long threshold,long[] numbers) {
    		this.start = start;
    		this.end = end;
    		this.threshold = threshold;
    		this.numbers = numbers;
    	}
    
    	@Override
    	protected Long compute() {
    		long sum = 0;
    		if((this.end - this.start) < threshold){
    			for(long i = this.start; i < this.end; i++){
    				sum += numbers[(int) i];
    			}
    		}else{
    			System.out.println("fork");//输出多少次，就表示有多少层
    			Long middle = (start + end) / 2;
    			Sum left = new Sum(start, middle, threshold, numbers);
    			Sum right = new Sum(middle, end, threshold, numbers);
    			left.fork();
    			right.fork();
    			sum = left.join() + right.join();
    		}
    		return sum;
    	}
    
    }

package com.np.ota.test.forkjoin2;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    
    public class Main {
    
    	public static void main(String[] args) {
    
    		ForkJoinPool pool = new ForkJoinPool();
    		long[] numbers = new long[30];
    		for (int i = 0; i < numbers.length; i++) {
    			numbers[i] = 1;
    		}
    		long threshold = 10;
    		long start = 0;
    		long end = numbers.length;
    		try {
    			ForkJoinTask<Long> result = pool.submit(new Sum(start, end,
    					threshold, numbers));
    			System.out.println(result.get());
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		} catch (ExecutionException e) {
    			e.printStackTrace();
    		}
    
    	}
    
    }

结果：

fork  
        fork  
        fork  
        30

参考：[https://blog.csdn.net/timheath/article/details/71307834][https_blog.csdn.net_timheath_article_details_71307834]

[https_blog.csdn.net_timheath_article_details_71307834]: https://blog.csdn.net/timheath/article/details/71307834