Java多线程/线程池

总结

执行器(Executor, ExecutorService)，
线程池(ThreadPoolExecutor, ScheduledThreadPoolExecutor, ForkJoinPool)，
线程，线程创建工厂
任务队列(无界队列-LinkedBlockingQueue, 同步队列-SynchronousQueue, 延迟队列-DelayedWorkQueue)，
任务(Runnable, Callable-FutureTask)，任务拒绝策略
关闭线程池

5个以上线程时使用线程池
alibaba要求自定义线程池，不适用Executors工具类

使用/自定义线程池new ThreadPoolExecutor()

     
    public static void main(String[] args) {
        
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                Runtime.getRuntime().availableProcessors(),
                Runtime.getRuntime().availableProcessors() * 2,
                30,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread th = new Thread(r, "threadName");
                        if (th.getPriority() != Thread.NORM_PRIORITY)
                            th.setPriority(Thread.NORM_PRIORITY);
                        return th;
                    }
                },
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        // 日志
                        System.out.println("任务" + r.toString() + "被拒绝");
                    }
                });

        for (int i = 1; i < 10; i++) {
            threadPoolExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(1000);
                        System.out.println("线程：" + Thread.currentThread().getName() + " 执行耗时任务");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }

        // 主线程执行任务
        for (int i = 0; i < 1; i++) {
            try {
                Thread.sleep(1000);
                System.out.println("主线程执行任务" + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 关闭线程池，否则线程池会维持核心线程数一直运行
        threadPoolExecutor.shutdown();
    }

使用/Executors/newSingleThreadExecutor/execute(Runnable command)无返回值

    // Executors.newSingleThreadExecutor/execute(Runnable command)
    public static void main(String[] args) {
        Runnable duty = new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++) {
                    try {
                        Thread.sleep(1000);
                        System.out.println("Runnable执行任务" + i);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        // execute(Runnable command) 只能传入Runnable
        executorService.execute(duty);
        // 主线程执行任务
        for (int i = 0; i < 10; i++) {
            try {
                Thread.sleep(1000);
                System.out.println("main执行任务" + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 关闭线程池，否则线程池会维持核心线程数一直运行
        executorService.shutdown();
    }

使用/Executors/newSingleThreadExecutor/submit(Callable task)有返回值

	// Executors.newSingleThreadExecutor/submit(Callable task)
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Callable duty = new Callable() {
            @Override
            public String call() {
                for (int i = 0; i < 10; i++) {
                    try {
                        Thread.sleep(1000);
                        System.out.println("Callable执行任务" + i);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                return "Callable任务执行完成";
            }
        };
        
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        
        Future future = executorService.submit(duty);
        // 主线程执行任务
        for (int i = 0; i < 3; i++) {
            try {
                Thread.sleep(1000);
                System.out.println("主线程执行任务" + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("主线程执行完任务");
        LocalDateTime start = LocalDateTime.now();
        // futureTask.get()会阻塞调用线程，这里阻塞主线程
        System.out.println("查看Callable线程执行结果：" + future.get());
        LocalDateTime end = LocalDateTime.now();
        Duration duration = Duration.between(start, end);
        System.out.println("主线程运行完毕，主线程被阻塞时间ms：" + duration.toMillis());
        // 关闭线程池，否则线程池会维持核心线程数一直运行
        executorService.shutdown();
    }

使用/Executors/newFixedThreadPool(nThread)

    public static void main(String[] args) {
        
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 3; i++) {
            Runnable duty = new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(3000);
                        System.out.println("线程：" + Thread.currentThread().getName() + " 执行任务");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            executorService.execute(duty);
        }

        // 主线程执行任务
        for (int i = 0; i < 3; i++) {
            try {
                Thread.sleep(1000);
                System.out.println("主线程执行任务" + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 关闭线程池，否则线程池会维持核心线程数一直运行
        executorService.shutdown();
    }

使用/Executors/newCachedThreadPool()

    public static void main(String[] args) {
        
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 3; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(3000);
                        System.out.println("线程：" + Thread.currentThread().getName() + " 执行任务");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }

        // 主线程执行任务
        for (int i = 0; i < 3; i++) {
            try {
                Thread.sleep(1000);
                System.out.println("主线程执行任务" + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 关闭线程池，否则线程池会维持核心线程数一直运行
        executorService.shutdown();
    }

使用/Executors/newScheduledThreadPool(corePoolSize)/固定间隔周期性执行scheduleAtFixedRate()

    public static void main(String[] args) {
        
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);
        
        for (int i = 0; i < 3; i++) {
            scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(1000);
                        System.out.println("线程：" + Thread.currentThread().getName() + " 执行任务");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, 1, 2, TimeUnit.SECONDS);
        }

        // 主线程执行任务
        for (int i = 0; i < 3; i++) {
            try {
                Thread.sleep(1000);
                System.out.println("主线程执行任务" + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 关闭线程池，否则线程池会维持核心线程数一直运行
        scheduledExecutorService.shutdown();
    }

使用/Executors/newScheduledThreadPool(corePoolSize)/固定延迟周期性执行scheduleWithFixedDelay()

    public static void main(String[] args) {
        
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);
        
        for (int i = 0; i < 3; i++) {
            scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
                @Override
                public void run() {
                    System.out.println("线程：" + Thread.currentThread().getName() + "开始执行任务时间" + LocalDateTime.now().getSecond());
                    try {
                        Thread.sleep(1000);
                        System.out.println("线程：" + Thread.currentThread().getName() + " 执行任务");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, 1, 1, TimeUnit.SECONDS);
        }

        // 主线程执行任务
        for (int i = 0; i < 5; i++) {
            try {
                Thread.sleep(1000);
                System.out.println("主线程执行任务" + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 关闭线程池，否则线程池会维持核心线程数一直运行
        scheduledExecutorService.shutdown();
    }

使用/Executors/newScheduledThreadPool(corePoolSize)/延迟执行1次schedule()

    public static void main(String[] args) {
        
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);
        
        System.out.println("当前时间：" + LocalDateTime.now().getSecond());
        for (int i = 0; i < 3; i++) {
            scheduledExecutorService.schedule(new Runnable() {
                @Override
                public void run() {
                    System.out.println("线程：" + Thread.currentThread().getName() + "开始执行任务时间" + LocalDateTime.now().getSecond());
                    try {
                        Thread.sleep(1000);
                        System.out.println("线程：" + Thread.currentThread().getName() + " 执行任务");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, 2, TimeUnit.SECONDS);
        }

        // 主线程执行任务
        for (int i = 0; i < 5; i++) {
            try {
                Thread.sleep(1000);
                System.out.println("主线程执行任务" + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 关闭线程池，否则线程池会维持核心线程数一直运行
        scheduledExecutorService.shutdown();
    }

使用/Executors.newWorkStealingPool()

newWorkStealingPool()本质上是ForkJoinPool
工作窃取算法：在线程池中还有线程可以提供服务的时候帮忙分担一些已经被分配给某一个线程的耗时任务
适合于有大量耗时任务的情况
ForkJoinPool线程池中的线程是守护线程，主线程任务执行完毕，所有守护线程都将停止

    public static void main(String[] args) {
        
        ExecutorService executorService = Executors.newWorkStealingPool();
        for (int i = 0; i < 5; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(1000);
                        System.out.println("线程：" + Thread.currentThread().getName() + " 执行耗时任务");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        // 主线程执行任务
        for (int i = 0; i < 1; i++) {
            try {
                Thread.sleep(1000);
                System.out.println("主线程执行任务" + i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 关闭线程池，否则线程池会维持核心线程数一直运行
        executorService.shutdown();
    }

使用/shutdown()

ExecutorService executorService = Executors.newSingleThreadExecutor();
// 停止ExecutorService添加新的任务, 继续执行正在执行的任务.
executorService.shutdown();

使用/shutdownNow()

ExecutorService executorService = Executors.newSingleThreadExecutor();
// 立即停止线程池，正在执行的任务也就关闭
executorService.shutdownNow();

理论/使用时机

5个以上线程时使用线程池

理论/运行流程

核心线程数》任务队列满》最大线程数》任务拒绝策略；只有在队列满情况下才会创建大于corePoolSize的线程

理论/线程池分类

ThreadPoolExecutor：

Executors.newSingleThreadExecutor()
Executors.newFixedThreadPool(3)
Executors.newCachedThreadPool()

ScheduledThreadPoolExecutor

Executors.newScheduledThreadPool(3)

ForkJoinPool

Executors.newWorkStealingPool()

理论/阻塞队列

BlockingQueue

线程安全的队列

ArrayBlockingQueue implements BlockingQueue

LinkedBlockingQueue implements BlockingQueue

newSingleThreadExecutor内部使用ThreadPoolExecutor+无界队列LinkedBlockingQueue；
newFixedThreadPool内部使用ThreadPoolExecutor+无界队列LinkedBlockingQueue；

SynchronousQueue implements BlockingQueue

newCachedThreadPool内部使用ThreadPoolExecutor+同步队列SynchronousQueue；
capacity是0，SynchronousQueue不存储元素，只起到同步和中转的作用
synchronousQueue.put(object)，synchronousQueue.take()，二者互为前提，类似互斥信号量

DelayQueue implements BlockingQueue

用于放置实现了Delayed接口的对象，其中的对象只能在其到期时才能从队列中取走。
DelayQueue是一个无界的BlockingQueue，用于放置实现了Delayed接口的对象，其中的对象只能在其到期时才能从队列中取走。这种队列是有序的，即队头对象的延迟到期时间最长。注意：不能将null元素放置到这种队列中。

DelayedWorkQueue implements BlockingQueue

newScheduledThreadPool内部使用ScheduledThreadPoolExecutor +延迟队列DelayedWorkQueue；
DelayedWorkQueue作用类似DelayQueue，差异在ScheduledFutureTask会记录延迟任务的索引

理论/线程数量

线程数(CPU密集型)=CPU核心数的1到2倍
线程数(I/O密集型)=CPU核心数 × (1+平均等待时间/平均工作时间)，IO时间越长线程数量越大

理论/任务拒绝策略

AbortPolicy（中断策略）：

抛出异常的策略拒绝新任务
中断最终都是抛出InterruptedException

DiscardPolicy（丢弃策略）：

丢弃任务的策略拒绝新任务
丢弃之后不会通知

DiscardOldestPolicy（丢弃最老的）：

丢弃最老任务的策略接收新任务
任务队列中存储了很多任务，丢弃在队列中存在时间最久的任务，以接收新的任务

CallerRunsPolicy（调用者运行）：

调用者执行的策略执行新任务，谁调用谁执行
例如，主线程给线程池提交任务，但是线程池已经满了，这种策略下会让提交任务的主线程去执行其提交的任务

理论/创建线程池

构造方法入参：核心线程数、最大线程数、存活时间、任务队列

Executors.newSingleThreadExecutor()

newSingleThreadExecutor内部使用ThreadPoolExecutor+无界队列LinkedBlockingQueue；
1=核心线程数=最大线程数

Executors.newFixedThreadPool(10)

newFixedThreadPool内部使用ThreadPoolExecutor+无界队列LinkedBlockingQueue；
核心线程数=最大线程数

Executors.newCachedThreadPool()

newCachedThreadPool内部使用ThreadPoolExecutor+同步队列SynchronousQueue；
核心线程数=0，最大线程数=Integer.MAX_VALUE

Executors.newScheduledThreadPool(5)

newScheduledThreadPool内部使用ScheduledThreadPoolExecutor +延迟队列DelayedWorkQueue；
最大线程数=Integer.MAX_VALUE

Executors.newWorkStealingPool(5)

Executors.newWorkStealingPool(5)内部使用ForkJoinPool+没有队列

自定义线程池：

ThreadPoolExecutor,
ScheduledThreadPoolExecutor,
ForkJoinPool

理论/停止线程池

shutdown()：停止ExecutorService添加新的任务, 但是老任务还是会继续执行.
isShutdown()：
isTerminated()：

问题/OOM

无界队列LinkedBlockingQueue，因为当处理任务的速度赶不上任务提交的速度的时候会造成OOM问题
无界线程池newCachedThreadPool、newScheduledThreadPool，因为线程的无限创建而导致OOM问题

问题/Executors.newSingleThreadExecutor的new LinkedBlockingQueue(){this(Integer.MAX_VALUE);}

因为：Executors.newSingleThreadExecutor的corePoolSize=1, maximumPoolSize=1，线程池中只有1个线程，因此任务来了只能放到任务队列中。

问题/为什么叫固定线程池 newFixedThreadPool

因为：ExecutorService executorService = Executors.newFixedThreadPool(3);的corePoolSize=3=maximumPoolSize=3，核心线程数和最大线程数相等

问题/为什么CachedThreadPool使用SynchronousQueue

因为：-核心线程数=0，最大线程数=Integer.MAX_VALUE，新任务来了将直接创建新线程执行这个新任务，使用SynchronousQueue队列中转新任务，新线程创建完成后将马上执行新任务，不会在任务队列中存储很久，因此使用同步队列中转一会儿新任务即可

问题/为什么CachedThreadPool的corePoolSize=0，maximumPoolSize=Integer.MAX_VALUE

因为：CachedThreadPool使用同步队列SynchronousQueue中转任务，强调任务能够被新的线程执行

问题/为什么ScheduledThreadPool使用DelayedWorkQueue

因为：调度线程池中线程需要执行的任务是定时任务，会在队列中存储一段时间，因此使用延迟队列DelayedWorkQueue

Java多线程/线程池

Java相关栏目本月热门文章