JDK并发包（线程池）（2）

短命女 2022-05-17 08:10 210阅读 0赞

在线程池中寻找堆栈

首先来看下这段代码

package thread.pool;
    
    import java.util.concurrent.SynchronousQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class Demo {
    	public static class DivTask implements Runnable{
    
    		int a,b;
    		public DivTask(int a,int b) {
    			this.a=a;
    			this.b=b;
    		}
    		
    		@Override
    		public void run() {
    			double re=a/b;
    			System.out.println(re);
    		}
    		
    	}
    	public static void main(String[] args) {
    		ThreadPoolExecutor pool=new ThreadPoolExecutor(0, Integer.MAX_VALUE, 
    				0, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
    		
    		for(int i=0;i<5;i++) {
    			pool.submit(new DivTask(100,i));
    		}
    		
    	}
    	
    }

讲道理啊，他会输出5次，可是实际情况是：

![70][]

这组程序只有4组数据 并且没有报任何错误信息。一种最简单的方法是将submit()改为execute()。或者使用Future。

![70 1][]

但是这两种方法只能得到部分信息。并不能得到详细的行数。

为了能打印详细的堆栈信息所以重写了ThreadPoolExecutor类：

package thread.pool;
    
    import java.util.concurrent.*;
    
    /**
     * 自己扩展线程池获取异常位置 重写submit:可以拉取异常
     */
    public class TraceThreadPoolExecutor extends ThreadPoolExecutor {
    
    	// 初始化
    	public TraceThreadPoolExecutor(int corePoolSize, int maxmumPoolSize, long keepAliveTime, TimeUnit unit,
    			BlockingQueue<Runnable> workQueue) {
    		super(corePoolSize, maxmumPoolSize, keepAliveTime, unit, workQueue);
    	}
    
    	// 执行方法
    	@Override
    	public void execute(Runnable task) {
    		super.execute(wrap(task, clientTrace(), Thread.currentThread().getName()));
    	}
    
    	@Override
    	public Future<?> submit(Runnable task) {
    		return super.submit(wrap(task, clientTrace(), Thread.currentThread().getName()));
    	}
    
    	private Exception clientTrace() {
    		// 扔出异常
    		return new Exception("堆栈跟踪");
    	}
    
    	private Runnable wrap(final Runnable task, final Exception clientStack, String clientThreadName) {
    
    		return new Runnable() {
    			public void run() {
    
    				try {
    					task.run();
    				} catch (Exception e) {
    					clientStack.printStackTrace();
    					throw e;
    				}
    			}
    		};
    	}
    
    	public static class DivTask implements Runnable {
    
    		int a, b;
    
    		public DivTask(int a, int b) {
    			this.a = a;
    			this.b = b;
    		}
    
    		@Override
    		public void run() {
    			double re = a / b;
    			System.out.println(re);
    		}
    
    	}
    
    	public static void main(String[] args) {
    		ThreadPoolExecutor pools = new TraceThreadPoolExecutor(0, Integer.MAX_VALUE, 0l, TimeUnit.SECONDS,
    				new SynchronousQueue<Runnable>());
    		// 错误堆栈中可以看到是在哪里提交的任务
    		for (int i = 0; i < 5; i++) {
    			pools.execute(new DivTask(100, i));
    		}
    	}
    }

![70 2][]

在重写的wrap()方法的第二个参数是一个异常，里面保存提交任务的线程堆栈信息，该方法将我们传入的Runnable任务进行了一层包装使其能处理异常信息。当任务发生异常时，这个异常会被打印。

Fork/Join框架

“分而治之”是非常有效地处理大量数据的方法。在Linux平台函数fork()用来创建子进程，使系统进程可以 多一个执行分支。在Java中也沿用了类似的命名方式。

join()的含义是表示等待。也就是使用fork()后系统多了一个执行分支（线程），所以需要等待这个执行分支执行完毕才能得到最终结果，所以join表示等待。

在实际使用中如果毫无顾忌使用fork()开启线程进行处理，那么很有可能导致系统开启过多的线程而严重影响性能。所以在JDK中给出了一个ForkJoinPool线程池，它不会急着开启线程，而是提交给ForkJoinPool线程池进行处理以节省系统资源。

![70 3][]

而由于线程的优化，提交的任务和线程的数量并不是一对一的关系。在大多数情况下，一个物理线程实际上需要处理多个逻辑任务。因此每个线程都拥有一个任务队列。当一个线程执行完自己的任务后会帮助其他线程执行任务，它会从底部开始拿数据，而当前线程会从顶部开始拿，这种机制有效的避免了数据竞争。

![70 4][]

ForkJoinPool的一个重要接口：

public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task)

你可以向ForkJoinPool线程池提交一个ForkJoinTask任务（支持fork()分解以及join()等待的任务。）ForkJoinTask有两个重要的子类，RecursiveAction和RecusiveTask。他们分别表示没有返回值的任务和可以携带返回值得任务。

举个例子通过Fork/Join框架进行数列求和：

package thread.pool;
    
    import java.util.ArrayList;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    import java.util.concurrent.RecursiveTask;
    
    public class CountTask extends RecursiveTask<Long> {
        private static final int THRESHOLD=1000;
        private long start;
        private long end;
    
    
        public CountTask(long start, long end) {
            super();
            this.start = start;
            this.end = end;
        }
    
    
        @Override
        protected Long compute() {
            long sum=0;
            boolean canCompute=(end - start)<THRESHOLD;
            if(canCompute){
                for (long i = start; i < end; i++) {
                    sum+=i;
                }
    
            }else{
                long step=(end - start)/100;
                ArrayList<CountTask> sunTasks = new ArrayList<CountTask>();
                long pos=start;
                for (int i = 0; i <100; i++) {
                    long lastOne=pos+step;
                    if(lastOne>end){
                        lastOne=end;
                    }
                    CountTask sunTask=new CountTask(pos,lastOne);
                    pos+=step+1;
                    sunTasks.add(sunTask);
                    sunTask.fork();
                }
                for (CountTask t : sunTasks) {
                    sum+=t.join();
                }
    
    
            }
    
            return sum;
        }
    
    
        public static void main(String[] args) throws InterruptedException, ExecutionException {
            ForkJoinPool forkJoinPool = new ForkJoinPool();
    
            for(int i=1;i<10000;i++) {
            	CountTask countTask = new CountTask(0,i);
            	
            	ForkJoinTask<Long> result = forkJoinPool.submit(countTask);
            	
            	Long res = result.get();
            	System.out.println(res);
            	
            }
    
    
    
        }
    
    }

![70 5][]

因为需要返回值所以这里用的是RecursiveTask作为任务模型。ForkJoinPool forkJoinPool = new ForkJoinPool();构建一个ForkJoinPool线程池，CountTask countTask = new CountTask(0,i);构造了了一个计算0~i求和的任务。然后通过ForkJoinTask<Long> result = forkJoinPool.submit(countTask);将任务提交。如果 boolean canCompute=(end - start)<THRESHOLD;为true表示小于1000则用for循环直接相加。否则将其分解，每次分解时简单的先将原有任务划分成100个等规模的小任务并使用fork()提交子任务，最通过循环那个列表join获取最后的值。

![70 6][]

需要注意的是如果任务划分层次很深，一直没有返回，那么可能出现两种情况，第一：线程数越来越多导致性能严重下降。第二：函数的调用层次变得很深最终导致堆栈溢出。

此外ForkJoin线程池使用一个无锁的栈来管理空闲线程，如果一个工作线程暂时取不到可用的任务，则可能会被挂起，挂起的线程会被压入由线程池维护的栈中。待将来有任务可用时，再从栈中唤醒这些线程。

参考《实战java高并发程序设计》

[70]: /images/20220517/009b4fcb4cbd43178a0f90e51c012e95.png
[70 1]: /images/20220517/9326d7fc065b4a96b2f2894ad2643bf0.png
[70 2]: /images/20220517/131f1a9cf65a468e8442d091729b66b0.png
[70 3]: /images/20220517/d80311bad1de4727bcc5722ad4eae9b7.png
[70 4]: /images/20220517/514f1d416cba4b13aa2d108c2b701200.png
[70 5]: /images/20220517/fbf231a51d9d4e3e92da06c8230f35b8.png
[70 6]: /images/20220517/1d9579cd25714027924febb44073e7ca.png