多线程——个人测试及理解

ゝ一纸荒年。 2023-06-21 12:27 9阅读 0赞

*  1、继承Thread类
 *  2、实现Runnable接口
 *  3、使用线程池，ExecutorService,Callable,Future实现（jdk1.5以上）
 *  4、使用线程池，ForkJoinPool实现（jdk1.7以上）

**下面，以计算1~1亿加和为例，测试以上4种多线程写法**

> 一、继承Thread类，重写run方法

package com.hky.syn;
    
    import java.time.Duration;
    import java.time.Instant;
    import java.util.Arrays;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    import java.util.stream.LongStream;
    
    public class SynThread {
    	private static long sum = 0;
    	private static Lock lock = new ReentrantLock();;
    	
    	public static void main(String[] args) {
    		long result = 0;
    		
    		long[] numbers = LongStream.rangeClosed(1, 100000000).toArray();
    		Instant now = Instant.now();
    		
    		long[] arr1 = Arrays.copyOf(numbers, numbers.length/2);
    		long[] arr2 = Arrays.copyOfRange(numbers, numbers.length/2, numbers.length);
    		System.out.println(arr1.length);
    		System.out.println(arr2.length);
    		
    		SynThreadSon son1 = new SynThreadSon(arr1);
    		SynThreadSon son2 = new SynThreadSon(arr2);
    		
    		Thread th1 = new Thread(son1);
    		Thread th2 = new Thread(son2);
    		th1.start();
    		th2.start();
    		
    		Instant end = Instant.now();
    		System.out.println("耗时：" + Duration.between(now, end).toMillis() + "ms");
    		System.out.println("结果为：" + result);
    	}
    	
    	static class SynThreadSon extends Thread{
    		private long arr[];
    		public SynThreadSon(long[] arr){
    			this.arr = arr;
    		}
    		@Override
    		public void run() {
    			long res = 0;
    			for (int i = 0; i < arr.length; i++) {
    				res += arr[i];
    			}
    			
    			lock.lock();
    			sum += res;
    			System.out.println("总数："+sum);
    			lock.unlock();
    		}
    	}
    }

> 二、实现runable接口方式

将上面的内部类代码，改成实现runable就可以了，基本类似，不写了。

> 三、使用线程池，ExecutorService,Callable,Future实现

package com.hky.forkjoin.executor;
    
    import java.time.Duration;
    import java.time.Instant;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    import java.util.stream.LongStream;
    
    import com.hky.forkjoin.basic.MyCalculator;
    
    public class ExecutorServiceCalculator implements MyCalculator{
    
    	private int cpus;
    	private ExecutorService pool;
    	
    	public ExecutorServiceCalculator() {
    		cpus = Runtime.getRuntime().availableProcessors();// CPU的核心数 默认就用cpu核心数了
    		pool = Executors.newFixedThreadPool(cpus);
    	}
    	
    
    	
    	@Override
    	public long sumUp(long[] numbers) {
    		// 每组个数
    		int parts = numbers.length / cpus;	// 1000/4 = 250
    //		for (int offset = 0, total = numbers.length; offset < total; offset += parts) {
    //			
    //		}
    		List<Future<Long>> result = new ArrayList<Future<Long>>();
    		for (int i = 0; i < cpus; i++) {
    			int from = i * parts;	// 0， 250，500，1000
    			int to = (i == cpus - 1) ? numbers.length - 1 : (i + 1) * parts - 1;
    			System.out.println("from="+from+",to="+to);
    			
    			Future<Long> future = pool.submit(new SumTask(numbers, from, to));
    			result.add(future);
    		}
    		
    		long total = 0;
    		for (Future<Long> future : result) {
    			try {
    				total += future.get();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			} catch (ExecutionException e) {
    				e.printStackTrace();
    			}
    		}
    		pool.shutdown();
    		return total;
    	}
    	
    	// 处理计算任务的线程
    	static class SumTask implements Callable<Long>{
    
    		private long[] numbers;
    		private int from;
    		private int to;
    		
    		public SumTask(long[] numbers, int from, int to) {
    			this.numbers = numbers;
    			this.from = from;
    			this.to = to;
    		}
    		
    		@Override
    		public Long call() throws Exception {
    			long total = 0L;
    			for (int i = from; i <= to; i++) {
    				total += numbers[i];
    			}
    			return total;
    		}
    		
    	}
    	// 59ms
    	public static void main(String[] args) {
    		long[] numbers = LongStream.rangeClosed(1, 100000000).toArray();
    		Instant now = Instant.now();
    
    		MyCalculator calculator = new ExecutorServiceCalculator();
    		long result = calculator.sumUp(numbers);
    		
    		Instant end = Instant.now();
    		System.out.println("耗时：" + Duration.between(now, end).toMillis() + "ms");
    		System.out.println("结果为：" + result);
    	}
    }

> 四、使用线程池，ForkJoinPool实现

*  fork()：开启一个新线程（或是重用线程池内的空闲线程），将任务交给该线程处理。
 *  join()：等待该任务的处理线程处理完毕，获得返回值。
 *  执行任务RecursiveTask：有返回值 RecursiveAction：无返回值
 *  在compute方法中，对任务进行拆解的好坏，决定了效率的高低

需要注意的是，并不是fork一次就是开启一个新线程的。每个 join() 也不是一定会造成线程被阻塞。使用的是**work stealing算法（任务窃取算法）**。

参考：[https://blog.csdn.net/f641385712/article/details/83749798][https_blog.csdn.net_f641385712_article_details_83749798]

1.  ForkJoinPool 的每个工作线程都维护着一个工作队列（WorkQueue），这是一个**双端队列**（Deque），里面存放的对象是任务（ForkJoinTask）。
2.  每个工作线程在运行中产生新的任务（通常是因为调用了 fork()）时，会**放入工作队列的队尾**，并且工作线程在处理自己的工作队列时，使用的是 LIFO 方式，也就是说每次从**队尾取出任务**来执行。
3.  每个工作线程在处理自己的工作队列同时，会**尝试窃取一个任务**（或是来自于刚刚提交到 pool 的任务，或是来自于其他工作线程的工作队列），窃取的任务位于**其他线程的工作队列的队首**，也就是说工作线程在窃取其他工作线程的任务时，使用的是 FIFO 方式。
4.  在遇到 join() 时，如果需要 join 的任务尚未完成，则会先处理其他任务，并等待其完成。
5.  在既没有自己的任务，也没有可以窃取的任务时，进入**休眠**。

package com.hky.forkjoin;
    
    import java.time.Duration;
    import java.time.Instant;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.RecursiveTask;
    import java.util.stream.LongStream;
    
    import com.hky.forkjoin.basic.MyCalculator;
    
    /**
     * 计算1至10000000的正整数之和。
     * 
     * @author hky
     *
     */
    public class ForkjoinTaskTest implements MyCalculator{
    	
    	private ForkJoinPool pool;
    	
    	public ForkjoinTaskTest(){
    		pool = new ForkJoinPool(15);//线程池大小，如果不加我的是9
    	}
    	
    	// 执行任务RecursiveTask：有返回值  RecursiveAction：无返回值
    	private static class SumTask extends RecursiveTask<Long> {
    		/**
    		 * 
    		 */
    		private static final long serialVersionUID = 1L;
    		private long[] numbers;
            private int from;
            private int to;
            
            public SumTask(long[] numbers, int from, int to){
            	this.numbers = numbers;
            	this.from = from;
            	this.to = to;
            }
            
            // 此方法为ForkJoin的核心方法：对任务进行拆分  拆分的好坏决定了效率的高低
    		@Override
    		protected Long compute() {
    			if (to-from<6) {//如果改成100000效果会更好
    				long total = 0;
    				for (int i = from; i <= to; i++) {
    					total += numbers[i];
    				}
    				return total;
    			} else {
    				int middle = (from+to)/2;
    				
    				SumTask task1 = new SumTask(numbers, from, middle);
    				SumTask task2 = new SumTask(numbers, middle+1, to);
    				
    				task1.fork();
    				task2.fork();
    				return task1.join() + task2.join();
    			}
    		}
    	}
    
    	@Override
    	public long sumUp(long[] numbers) {
    		Long res = pool.invoke(new SumTask(numbers, 0, numbers.length-1));
    		System.out.println("线程池大小" + pool.getPoolSize());
    		pool.shutdown();
    		return res;
    	}
    	
    	public static void main(String[] args) {
    		long[] numbers = LongStream.rangeClosed(1, 100000000).toArray();
    
    		Instant now = Instant.now();
    
    		MyCalculator calculator = new ForkjoinTaskTest();
    		long result = calculator.sumUp(numbers);
    		
    		Instant end = Instant.now();
    
    		System.out.println("耗时：" + Duration.between(now, end).toMillis() + "ms");
    		System.out.println("结果为：" + result);
    	}
    }

[https_blog.csdn.net_f641385712_article_details_83749798]: https://blog.csdn.net/f641385712/article/details/83749798