Java异步编程之ThreadPoolExecutor

雨点打透心脏的1/2处 2022-11-18 01:59 252阅读 0赞

在《阿里巴巴java开发手册》中指出了线程资源必须通过线程池提供，不允许在应用中自行显示的创建线程，这样对线程的创建更加规范，可以合理控制开辟线程的数量；而线程池不允许使用Executors去创建，而要通过ThreadPoolExecutor方式，这一方面是由于jdk中Executor框架虽然提供了如newFixedThreadPool()、newSingleThreadExecutor()等创建线程池的方法，但都有其局限性，不够灵活；另外由于前面几种方法内部也是通过ThreadPoolExecutor方式实现，使用ThreadPoolExecutor有助于大家明确线程池的运行规则，创建符合自己的业务场景需要的线程池，避免资源耗尽的风险。

Java中的异步任务原理是把任务提交到一个线程中运行，如此就可以利用线程池，把任务提交给线程池中的线程去执行，合理利用资源。

我们知道Java Thread类执行run方法，执行任务没有返回值的Runnable类，有返回值的Callable类，但是Thread不能直接执行Callable，需要通过FutureTask

//1.没有返回值Runnable
    new Thread(new Runnable(){ 
      @Override
      public void run() { 
         System.out.println("runnable...");
      }
    }).start();
    
    //2.有返回值Callable
    Callable callable = new Callable<String>() { 
        @Override
        public String call() throws Exception { 
            return "callable...";
        }
    };
    FutureTask<String> futureTask = new FutureTask<>(callable);
    new Thread(futureTask).start();
    System.out.println(futureTask.get());

当有了上面的基础知识后，我们只要把Runnable或Callable提交到线程池去执行即可，具体的参考如下Util工具类

public class AsyncTaskUtil { 
    
        private static final int corePoolSize = Runtime.getRuntime().availableProcessors();//CPU核心数
        private static final int maximumPoolSize = corePoolSize * 2;
        private static long keepAliveTime = 1;
        private static TimeUnit keepAliveTimeUnit = TimeUnit.MINUTES;
        private static BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(1024);
        private static RejectedExecutionHandler rejectedHandler = new ThreadPoolExecutor.CallerRunsPolicy();
    
        /** * corePoolSize : 线程池核心线程数，最好默认CPU核心数 * maximumPoolSize : 线程池最大线程数，最好是核心线程数的两倍，太多会引起线程切换 * keepAliveTime : 大于核心线程数的空闲线程存活时间 * keepAliveTimeUnit : 空闲线程存活时间的单位（秒、分钟、小时等等） * workQueue : 线程池有界队列，新任务没有可用线程处理时会把任务放到该队列中，等待被处理 * rejectedHandler : 拒绝处理策略，默认直接丢弃并抛出异常-AbortPolicy，调用者线程直接处理-CallerRunsPolicy */
        private static ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                keepAliveTimeUnit,
                workQueue,
                rejectedHandler
        );
    
        /** * 直接提交任务，无返回值 * @param task */
        public static void submit(Runnable task){ 
            poolExecutor.submit(task);
        }
    
        /** * 提交任务，返回泛型结果 * @param task * @param <T> * @return */
        public static <T> Future<T> submit(Callable<T> task){ 
            return poolExecutor.submit(task);
        }
    	
    	//测试
        public static void main(String[] args) throws Exception{ 
            long start = System.currentTimeMillis();
            //提交任务
            AsyncTaskUtil.submit(() -> ma());
            System.out.println("exec ma time : " + (System.currentTimeMillis() - start));
    		
    		//提交任务，获取执行结果
            Future<Integer> futureC = AsyncTaskUtil.submit(() -> mc());
            System.out.println(futureC.get());
            System.out.println("exec mc time : " + (System.currentTimeMillis() - start));
        }
    
        public static void ma(){ 
            int result = 1 + 1;
            System.out.println("ma 1 + 1 = " + result);
        }
        public static int mc(){ 
            int result = 1 + 3;
            System.out.println("ma 1 + 3 = " + result);
            return result;
        }
    }