-
Java多线程/线程池
总结
执行器(Executor, ExecutorService)，
线程池(ThreadPoolExecutor, ScheduledThreadPoolExecutor, ForkJoinPool)，
线程，线程创建工厂
任务队列(无界队列-LinkedBlockingQueue, 同步队列-SynchronousQueue, 延迟队列-DelayedWorkQueue)，
任务(Runnable, Callable-FutureTask)，任务拒绝策略
关闭线程池5个以上线程时使用线程池
alibaba要求自定义线程池，不适用Executors工具类使用/自定义线程池new ThreadPoolExecutor()
/** * 自定义线程池（阿里推荐） * 1. 明确使用什么线程池 * 2. 得到线程数：明确I/O密集型还是CPU密集型 * 3. 确定任务队列 * 4. 线程工厂 * 5. 任务拒绝策略 ** 线程数(CPU密集型)=CPU核心数的1到2倍 * 线程数(I/O密集型)=CPU核心数 × (1+平均等待时间/平均工作时间)，IO时间越长线程数量越大 * java获取CPU核心数：Runtime.getRuntime().availableProcessors() */
public static void main(String[] args) { /** * corePoolSize：核心线程数，线程池维持的线程数 * maximumPoolSize：池中最多线程数 * keepAliveTime：大于核心线程数的线程存活时间 * unit：存活时间单位 * workQueue：任务队列 * threadFactory：线程工厂 * RejectedExecutionHandler：任务拒绝策略处理器 */ 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(); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
使用/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(); } } } }; /** * new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()) * corePoolSize=1, maximumPoolSize=1, keepAliveTime=0L, TimeUnit.MILLISECONDS * new LinkedBlockingQueue 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(); }(){this(Integer.MAX_VALUE);} */ - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
使用/Executors/newSingleThreadExecutor/submit(Callable task)有返回值
// Executors.newSingleThreadExecutor/submit(Callabletask) public static void main(String[] args) throws ExecutionException, InterruptedException { Callable<String> duty = new Callable<String>() { @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任务执行完成"; } }; /** * new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()) * corePoolSize=1, maximumPoolSize=1, keepAliveTime=0L, TimeUnit.MILLISECONDS * new LinkedBlockingQueue ExecutorService executorService = Executors.newSingleThreadExecutor(); /** * submit(Callable(){this(Integer.MAX_VALUE);} */ task) * submit(Runnable task, T result) * submit(Runnable task) */ Future<String> 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(); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
使用/Executors/newFixedThreadPool(nThread)
public static void main(String[] args) { /** * new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()) * corePoolSize=3, maximumPoolSize=3, keepAliveTime=0L, TimeUnit.MILLISECONDS * new LinkedBlockingQueue 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(); }(){this(Integer.MAX_VALUE);} */ - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
使用/Executors/newCachedThreadPool()
public static void main(String[] args) { /** * new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS, new SynchronousQueue()) * corePoolSize=0, maximumPoolSize=Integer.MAX_VALUE, keepAliveTime=60L, TimeUnit.SECONDS * new SynchronousQueue 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(); }(){this(false);} true-公平，线程FIFO获取任务，false-不公平，线程争抢获取任务 */ - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
使用/Executors/newScheduledThreadPool(corePoolSize)/固定间隔周期性执行scheduleAtFixedRate()
public static void main(String[] args) { /** * new ScheduledThreadPoolExecutor(corePoolSize) * corePoolSize=3, maximumPoolSize=Integer.MAX_VALUE, keepAliveTime=0L, NANOSECONDS * new DelayedWorkQueue(){this(false);} true-公平，线程FIFO获取任务，false-不公平，线程争抢获取任务 */ ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3); /** * scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) * initialDelay：初始延迟时间 * period：执行间隔 * 执行耗时>执行间隔：任务执行完成后立马开始执行下一轮任务 * 执行耗时<等待时间：在任务执行完成后等待(等待时间-执行耗时)，再执行下一轮任务 */ 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(); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
使用/Executors/newScheduledThreadPool(corePoolSize)/固定延迟周期性执行scheduleWithFixedDelay()
public static void main(String[] args) { /** * new ScheduledThreadPoolExecutor(corePoolSize) * corePoolSize=3, maximumPoolSize=Integer.MAX_VALUE, keepAliveTime=0L, NANOSECONDS * new DelayedWorkQueue(){this(false);} true-公平，线程FIFO获取任务，false-不公平，线程争抢获取任务 */ ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3); /** * scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) * initialDelay：初始延迟时间 * delay：延迟时间 * 任务执行完成之后延迟delay时间再执行下一轮 */ 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(); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
使用/Executors/newScheduledThreadPool(corePoolSize)/延迟执行1次schedule()
public static void main(String[] args) { /** * new ScheduledThreadPoolExecutor(corePoolSize) * corePoolSize=3, maximumPoolSize=Integer.MAX_VALUE, keepAliveTime=0L, NANOSECONDS * new DelayedWorkQueue(){this(false);} true-公平，线程FIFO获取任务，false-不公平，线程争抢获取任务 */ ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3); /** * schedule(Runnable command, long delay, TimeUnit unit) * delay：延迟时间 * 在给定的延迟时间后执行1次任务 */ 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(); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
使用/Executors.newWorkStealingPool()
newWorkStealingPool()本质上是ForkJoinPool
工作窃取算法：在线程池中还有线程可以提供服务的时候帮忙分担一些已经被分配给某一个线程的耗时任务
适合于有大量耗时任务的情况
ForkJoinPool线程池中的线程是守护线程，主线程任务执行完毕，所有守护线程都将停止public static void main(String[] args) { /** * new ForkJoinPool(Runtime.getRuntime().availableProcessors(), ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true) * ForkJoinPool线程池中的线程是守护线程，主线程任务执行完毕，所有守护线程都将停止 * new ForkJoinPool(Runtime.getRuntime().availableProcessors(), ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true) */ 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(); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
使用/shutdown()
ExecutorService executorService = Executors.newSingleThreadExecutor(); // 停止ExecutorService添加新的任务, 继续执行正在执行的任务. executorService.shutdown();- 1
- 2
- 3
- 4
使用/shutdownNow()
ExecutorService executorService = Executors.newSingleThreadExecutor(); // 立即停止线程池，正在执行的任务也就关闭 executorService.shutdownNow();- 1
- 2
- 3
- 4
理论/使用时机
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
-
相关阅读:
软件架构风格
Python进阶系列 - 17讲 函数
阿里云高庆瑞：如何快速搭建低成本高弹性的云上应用
【记录】深度学习环境配置（pytorch版）
Quick Pow: 如何快速求幂
纬度转地理易语言代码
泛微OA E-Office V10 OfficeServer 任意文件上传漏洞复现
QT -C++小写转换成大写-QPushButton-setGeometry-QLineEdit-信号与槽
eclipse汉化方法
C++运算符优先级一览表
- 原文地址：https://blog.csdn.net/baidu_35805755/article/details/125897377
