Java线程池的实现原理及其在业务中的最佳实践

execute(Runnable r):没有返回值，仅仅是把一个任务提交给线程池处理

submit(Runnable r):返回值为Future类型，当任务处理完毕后，通过Future的get()方法获取返回值时候，得到的是null

submit(Runnable r,Object result):返回值为Future类型，当任务处理完毕后，通过Future的get()方法获取返回值时候，得到的是传入的第二个参数result 

shutdown():关闭线程池，不接受新任务，但是等待队列中的任务处理完毕才能真正关闭

shutdownNow():立即关闭线程池，不接受新任务，也不再处理等待队列中的任务，同时中断正在执行的线程

setCorePoolSize(int corePoolSize):设置核心线程数
    
setKeepAliveTime(long time, TimeUnit unit):设置线程的空闲时间
    
setMaximumPoolSize(int maximumPoolSize):设置最大线程数
    
setRejectedExecutionHandler(RejectedExecutionHandler rh):设置拒绝策略
    
setThreadFactory(ThreadFactory tf):设置线程工厂

beforeExecute(Thread t, Runnable r):任务执行之前的钩子函数，这是一个空函数，使用者可以继承ThreadPoolExecutor后重写这个方法，实现其中的逻辑
    
afterExecute(Runnable r, Throwable t):任务执行之后的钩子函数，这是一个空函数，使用者可以继承ThreadPoolExecutor后重写这个方法，实现其中的逻辑

  public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) 

1.corePoolSize:  核心线程数

2.maximumPoolSize:  最大线程数

3.keepAliveTime:  线程的空闲时间

4.unit:  空闲时间的单位（秒、分、小时等等）

5.workQueue:  等待队列

6.threadFactory:  线程工厂

7.handler:  拒绝策略

ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异

ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常

ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）

ThreadPoolExecutor.CallerRunsPolicy：由调用线程直接处理该任务（可能为主线程Main），保证每个任务执行完毕

public class TestThreadPool {

    /**
     * 线程池
     */
    private static ExecutorService executor = initDefaultExecutor();

    /**
     * 统一的获取线程池对象方法
     */
    public static ExecutorService getExecutor() {
        return executor;
    }

    private static final int DEFAULT_THREAD_SIZE = 16;
    private static final int DEFAULT_QUEUE_SIZE = 10240;

    private static ExecutorService initDefaultExecutor() {
        return new ThreadPoolExecutor(DEFAULT_THREAD_SIZE, DEFAULT_THREAD_SIZE,
                300, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(DEFAULT_QUEUE_SIZE),
                new DefaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }
}

public static void main(String[] args) {
    test1();
    test2();
}

public static void test1(){

    Object obj = new Object();
    System.out.println("方法一执行完成");
}

public static void test2(){

    ExecutorService executorService = Executors.newFixedThreadPool(10);
    executorService.execute(new Runnable() {
        @Override
        public void run() {
            System.out.println("方法二执行完成");
        }
    });
}

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable

public class Outer {

    private String name;

    private Inner inner;

    public int outerMethod() {
        return 1;
    }

    /**
     * 非静态内部类
     */
    class Inner {

        private void innerMethod() {
            //在非静态内部类中可以直接调用外部类的方法
            outerMethod();
        }

        private String address;
    }
}

class Outer$Inner {
    private String address;

    Outer$Inner(Outer var1) {
        this.this$0 = var1;
    }

    private void innerMethod() {
        this.this$0.outerMethod();
    }
}

public class Outer {

    private String name;

    private Inner inner;

    public int outerMethod() {
        return 1;
    }

    /**
     * 静态内部类
     */
    static class Inner {

        private String address;
    }
}

class Outer$Inner {
    private String address;

    Outer$Inner() {
    }
}

public class FartherAndSonTask {

    public static ExecutorService executor= TestThreadPool.getExecutor();

    public static void main(String[] args) throws Exception {
        FatherTask fatherTask = new FatherTask();
        Future<String> future = executor.submit(fatherTask);
        future.get();
    }

    /**
     * 父任务，里面异步执行子任务
     */
    static class FatherTask implements Callable<String> {
        @Override
        public String call() throws Exception {
            System.out.println("开始执行父任务");
            SonTask sonTask = new SonTask();
            Future<String> future = executor.submit(sonTask);
            String s = future.get();
            System.out.println("父任务已拿到子任务执行结果");
            return s;
        }
    }
    /**
     * 子任务
     */
    static class SonTask implements Callable<String> {
        @Override
        public String call() throws Exception {

            //处理一些业务逻辑
            System.out.println("子任务执行完成");
            return null;
        }
    }
}

private void asyncSupplyPriceSync(List<Long> shidList, SupplyPriceSyncMsg msg) {

        if (CollectionUtils.isEmpty(shidList)) {
            return;
        }
        PlatformLogUtil.logInfo("异步推送酒店报价信息供给总数：", shidList.size());
        final Map<String, Future<?>> futures = Maps.newLinkedHashMap();
        //分批提交线程池处理
        Lists.partition(shidList, SwitchConfig.HOTEL_PRICE_ASYNC_LIST_SIZE)
                .forEach(subList -> {
                    try {
                        futures.put(UUID.randomUUID().toString(), executorService
                                .submit(() -> batchSupplyPriceSync(subList, msg)));
                    } catch (Exception e) {
                        PlatformLogUtil.logFail("异步推送报价信息线程池子任务执行异常", LogListUtil.newArrayList(subList), e);
                    }
                });
        //阻塞，等所有子任务都处理完，才返回结果
        futures.forEach((uuid, future) -> {
            try {
                future.get();
            } catch (InterruptedException | ExecutionException e) {
                PlatformLogUtil.logFail("异步推送报价信息获取子任务执行结果异常", LogListUtil.newArrayList(e));
            }
        });
    }

public class ExceptionTest {

    public static ExecutorService executor = TestThreadPool.getExecutor();

    public static void main(String[] args) {

        executor.execute(() -> test("正常"));
        executor.execute(() -> test("正常"));
        executor.execute(() -> test("任务执行异常"));
        executor.execute(() -> test("正常"));
        executor.shutdown();
    }

    public static void test(String str) {

        String result = "当前ThreadName为" + Thread.currentThread().getName() + ":结果" + str;
        if (str.equals("任务执行异常")) {
            throw new RuntimeException(result + "****执行异常");
        } else {
            System.out.println(result);
        }
    }
}